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 SearchRoomToolSchema(BaseModel): from_date: str = Field(description="Start date of booking. Format is MM-DD-YYYY.") to_date: datetime = Field(description="End date of booking. Format is MM-DD-YYYY.") adult: int = Field(description="Count of adult", default=2) child: int = Field(description="Count of child", default=0) # room_filters: list[str] = Field(description="User Provided Keywords on which we find rooms, like 'room_name', 'amenities', etc", default=None) def __init__(self,from_date:datetime, *,to_date:datetime = None, adult:int = 0, child:int = 0) -> None: self.from_date = from_date self.to_date = to_date self.adult = adult self.child = child def __setattr__(self, name, value) -> None: self.__dict__[name] = value class SearchRoomSchema(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.") room_ids:List[int] = Field(description="Enter the room_type_id of that rooms which you filtered") class Config: extra = Extra.allow # Allow extra properties arbitrary_types_allowed = True SEARCH_ROOM_PROMPT = """You are a helpful AI assistant trained to answer questions about "Bay View Hotel" -> Available Rooms Details. Use the following pieces of context 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. Today's in date DD/MM/YYYY for your Knowledge: {date} Include any relevant images or URLs, but only if they are found in the context. Don't make answer too much large for user gave you view in concise way (Follow law of least action). {context} Always list filtered rooms according to user question else list all rooms if no critaria set by user. Always add 'room_type_id' in your response While listing filtered rooms list according to user question in your response. 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]] = [ SystemMessagePromptTemplate.from_template(SEARCH_ROOM_PROMPT), *_prompts, HumanMessagePromptTemplate.from_template("{question}") ] return ChatPromptTemplate(messages=messages) class SearchRoomTool: 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="search_or_check_hotel_rooms_availabilities", description="Use this tool to Search or show or check hotels rooms availability with price. Only use this tool if all parameters are present in user query, if not then ask the user for required parameters to use this tool. Always use this tool before book_hotel_room tool.", func=self._search_rooms, args_schema=SearchRoomToolSchema, return_direct=True ) def _search_rooms(self, callbacks: CallbackManager = None, from_date = None, to_date = None, adult=0, child=0, **kwargs): chain = create_structured_output_runnable( llm=self.llm, output_schema=SearchRoomSchema, prompt=create_search_tool_prompt(extra_prompt_messages=self.chat_hisory) ) rooms_list = self._get_rooms(from_date, to_date, adult, child, **kwargs) context = rooms_list.__str__() if len(rooms_list) else "No Rooms Found" result: SearchRoomSchema = chain.invoke(input={'question': self.question, "context": context, "date": datetime.now().strftime('%m-%d-%Y')}, config={"callbacks":callbacks}) result.additional = None if len(rooms_list): param_available = True if adult and from_date and to_date else False result.additional = ChatAdditonal( actions=["change-url", "scrollTo"], # createDiv= "SHOW-ROOM-LIST", roomList=rooms_list, url= f"/search-results/{self.sub_hotel_id}/{(from_date)}/{(to_date)}/{adult}/{child}" if param_available else "/Suites", scrollToElement= ".search-result-inner" if param_available else ".room-listing-block" ) return result def _get_rooms(self, from_date = None, to_date = None, adult=0, child=0, **kwargs) -> list: try: response = requests.post( self.hotel_url + "frontend/Searchresults/get_search_results_sj", json={ "adult" : str(adult), "child" : str(child), "promo": "", "room_stay_from": from_date, "room_stay_to": to_date, "sub_hotel_id": "124", "type": "room" }, headers={"current_version": self.hotel_current_verion} ) rooms_list=[] rooms_data = json.loads(response.content.decode('utf-8-sig')) for items in rooms_data['data']['rooms_avail']: for item in items['available_rooms']: price = None for key,val in item["rates"]['2'].items(): price = sum(float(p) for p in val) rooms_list.append({ "roomName": item['room_display_name'], "roomImage": item['image'], "bundleRoom": item["is_bundle"], "occupancy":item["max_occupancy"], "bed": item["bed_code_name"], "room_type_code": item["product_id"], "room_type_id": item["type_id"], "rate": "{:.2f}".format(price) }) if response.status_code != 200 or rooms_data['status'] == 'error': raise Exception("Error Occur in While Room Fetching") 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)