import random from abc import ABC, abstractmethod from typing import get_origin, get_args import os # from langchain.tools import tool import json from pydantic import BaseModel, Field from typing import Dict, Union import random import copy from types import UnionType from langchain_community.vectorstores import FAISS from langchain_community.embeddings import HuggingFaceEmbeddings class VectorStore: def __init__(self, embeddings_model, vectorstore): embeddings = HuggingFaceEmbeddings(model_name=embeddings_model) self.vectore_store = FAISS.load_local(vectorstore, embeddings, allow_dangerous_deserialization=True) def get_context(self, instruction, number_of_contexts=2): documentos = self.vectore_store.similarity_search_with_score(instruction, k=number_of_contexts) return self._beautiful_context(documentos) def _beautiful_context(self, docs): context = "" for doc in docs: context += doc[0].page_content + "\n" return context def read_json(data_path: str) -> tuple[list, dict]: try: with open(data_path, 'r', encoding="utf-8") as f: data = [json.loads(line) for line in f.readlines()] except: with open(data_path, 'r', encoding="utf-8") as f: data = json.loads(f.read()) return data json_data = read_json("data/val_de_nuria.json") reservations = {} class ToolBase(BaseModel, ABC): @abstractmethod def invoke(cls, input: Dict): pass @classmethod def to_openai_tool(cls): """ Extracts function metadata from a Pydantic class, including function name, parameters, and descriptions. Formats it into a structure similar to OpenAI's function metadata. """ function_metadata = { "type": "function", "function": { "name": cls.__name__, # Function name is same as the class name, in lowercase "description": cls.__doc__.strip(), "parameters": { "type": "object", "properties": {}, "required": [], }, }, } # Iterate over the fields to add them to the parameters for field_name, field_info in cls.model_fields.items(): # Field properties field_type = "string" # Default to string, will adjust if it's a different type annotation = field_info.annotation.__args__[0] if getattr(field_info.annotation, "__origin__", None) is Union else field_info.annotation has_none = False if get_origin(annotation) is UnionType: # Check if it's a Union type args = get_args(annotation) if type(None) in args: has_none = True args = [arg for arg in args if type(None) != arg] if len(args) > 1: raise TypeError("It can be union of only a valid type (str, int, bool, etc) and None") elif len(args) == 0: raise TypeError("There must be a valid type (str, int, bool, etc) not only None") else: annotation = args[0] if annotation == int: field_type = "integer" elif annotation == bool: field_type = "boolean" # Add the field's description and type to the properties function_metadata["function"]["parameters"]["properties"][field_name] = { "type": field_type, "description": field_info.description, } # Determine if the field is required (not Optional or None) if field_info.is_required(): function_metadata["function"]["parameters"]["required"].append(field_name) has_none = True # If there's an enum (like for `unit`), add it to the properties if hasattr(field_info, 'default') and field_info.default is not None and isinstance(field_info.default, list): function_metadata["function"]["parameters"]["properties"][field_name]["enum"] = field_info.default if not has_none: function_metadata["function"]["parameters"]["required"].append(field_name) return function_metadata tools: Dict[str, ToolBase] = {} oitools = [] vector_store = VectorStore(embeddings_model="BAAI/bge-m3", vectorstore="data/vs") def tool_register(cls: BaseModel): oaitool = cls.to_openai_tool() oitools.append(oaitool) tools[oaitool["function"]["name"]] = cls # @tool_register class hotel_description(ToolBase): """Retrieves basic information about the hotel, such as its name, address, contact details, and overall description.""" @classmethod def invoke(cls, input: Dict) -> str: return """### **Nou Vall de Núria – Brief Description** Nestled in the stunning **Vall de Núria** in the Pyrenees, **Nou Vall de Núria** offers a perfect blend of comfort and adventure. Guests can enjoy breathtaking mountain views, premium accommodations, and excellent facilities, including an outdoor pool, gym, and sauna. The hotel features **two dining options**, serving traditional Catalan cuisine and refreshing drinks. Accommodations range from **cozy standard rooms to luxurious suites and fully equipped apartments**, catering to couples, families, and groups. For an unforgettable stay, guests can choose from **special packages**, including family-friendly stays, romantic getaways, ski adventures, and relaxation retreats. Outdoor enthusiasts can explore **hiking trails, ski slopes, and fishing spots** in the surrounding natural beauty. Whether for relaxation or adventure, **Nou Vall de Núria** promises a unique and memorable experience.""" @tool_register class get_hotal_general_information(ToolBase): """ Retrieves general information about a hotel, including its background, amenities, location, and nearby attractions based on the query. """ query: str = Field(description="User's question or query for retrieving the information.") @classmethod def invoke(cls, input: Dict) -> str: query = input.get("query", None) if not query: return "Missing required argument: query." # return "We are currently working on it. You can't use this tool right now—please try again later. Thank you for your patience!" return vector_store.get_context(query) @tool_register class packs(ToolBase): """Provides a list of available activity pack at the hotel.""" @classmethod def invoke(cls, input: Dict) -> str: return json_data["packs"] @tool_register class hotel_facilities(ToolBase): """Provides a list of available general facilities at the hotel, which could include amenities like a spa, pool, gym, conference rooms, etc.""" @classmethod def invoke(cls, input: Dict) -> str: return json_data["general_facilities"] @tool_register class restaurants_details(ToolBase): """Provides a list of available restaurants with their details.""" @classmethod def invoke(cls, input: Dict) -> str: """ Play a playlist by its name, starting with the first or a random song. """ return json_data["restaurants"] @tool_register class restaurant_details(ToolBase): """Retrieves detailed information about a specific restaurant in the hotel, including its menu, ambiance, operating hours, and special features.""" name: str = Field(default=[res["name"] for res in json_data["restaurants"]], description="Name of the resaturant") @classmethod def invoke(cls, input: Dict) -> str: """ Play a playlist by its name, starting with the first or a random song. """ instance = cls(**input) name = instance.name restaurante = [res for res in json_data["restaurants"] if res["name"] == name] if restaurante: return restaurante else: return f"We don't have any restaurante with the name: {name}" @tool_register class rooms_information(ToolBase): """ Provides a list of available hotel rooms, including brief descriptions of each room type. """ @classmethod def invoke(cls, input: Dict) -> str: return json_data["room_types"] @tool_register class check_room_availability(ToolBase): """ Checks if a specified room type is available between the provided check-in and check-out dates for a given number of guests. """ room_type: str = Field(default=list(json_data["room_types"].keys()), description="The type of room to check for availability.") reservation_start_date: str = Field(description="The check-in date for the reservation, formatted as 'YYYY-MM-DD' (e.g., '2025-04-01')") reservation_end_date: str = Field(description="The check-out date for the reservation, formatted as 'YYYY-MM-DD' (e.g., '2025-04-05')") guests: int = Field(description="The number of guests that will occupy the room.") @classmethod def invoke(cls, input: Dict) -> str: input["room_type"] = input.get("room_type", None) room_type = input.get("room_type", None) reservation_start_date = input.get("reservation_start_date", None) reservation_end_date = input.get("reservation_end_date", None) guests = input.get("guests", None) missing = [] if not room_type: missing.append("room_type") if not reservation_start_date: missing.append("reservation_start_date") if not reservation_end_date: missing.append("reservation_end_date") if not guests: missing.append("guests") if len(missing): value = ", ".join(missing) return f"Unable to check the room availability. The following required arguments are missing:{value}." instance = cls(**input) room_type = instance.room_type reservation_start_date = instance.reservation_start_date reservation_end_date = instance.reservation_end_date guests = instance.guests rooms = [room for room in json_data["accomodations"]["rooms"] if room_type in room["type"]] if len(rooms) == 0: return f"There is no room exists with room type {room_type}" rooms2 = [room for room in rooms if guests <= room["number_of_guests"]] if len(rooms2) == 0: max_guests = json_data["room_types"][room_type]["number_of_guests"] return f"The number of guest is superior then the availibilty, maximum is {max_guests}" return rooms2 @tool_register class make_reservation(ToolBase): """ Creates a new reservation for the hotel by booking a room of the specified type for the desired dates, and associating the booking with a user. """ user_name: str = Field(description="The name of user who is doing the reservation.") room_type: str = Field(default=list(json_data["room_types"].keys()), description="The type of room being reserved.") reservation_start_date: str = Field(description="The check-in date for the reservation, formatted as 'YYYY-MM-DD' (e.g., '2025-04-01')") reservation_end_date: str = Field(description="The check-out date for the reservation, formatted as 'YYYY-MM-DD' (e.g., '2025-04-05')") guests: int = Field(description="The total number of guests for the reservation. Must be a positive integer.") @classmethod def invoke(cls, input: Dict) -> str: input["room_type"] = input.get("room_type", None) room_type = input.get("room_type", None) reservation_start_date = input.get("reservation_start_date", None) reservation_end_date = input.get("reservation_end_date", None) guests = input.get("guests", None) user_name = input.get("user_name", None) missing = [] if not room_type: missing.append("room_type") if not reservation_start_date: missing.append("reservation_start_date") if not reservation_end_date: missing.append("reservation_end_date") if not guests: missing.append("guests") if not user_name: missing.append("user_name") if len(missing): value = ", ".join(missing) return f"Unable to complete the reservation. The following required arguments are missing:{value}." instance = cls(**input) room_type = instance.room_type reservation_start_date = instance.reservation_start_date reservation_end_date = instance.reservation_end_date guests = instance.guests user_name = instance.user_name.lower() rooms = [room for room in json_data["accomodations"]["rooms"] if room_type in room["type"]] if len(rooms) == 0: return f"There is no room exists with room type {room_type}" rooms2 = [room for room in rooms if guests <= room["number_of_guests"]] if len(rooms2) == 0: max_guests = json_data["room_types"][room_type]["number_of_guests"] return f"The number of guest is superior then the availibilty, maximum is {max_guests}" room = rooms2[random.randint(0, len(rooms2) - 1)] rand = int(random.randint(0,10000000)) while rand in reservations: rand = int(random.randint(0,10000000)) tmp_data = { "status": "Reserved", "room_number": room["room_number"], "room_type": room_type, "reservation_start_date": reservation_start_date, "reservation_end_date": reservation_end_date, "guests": guests, "reservation_id": rand, "user_name": user_name, } reservations[rand] = tmp_data return json.dumps(tmp_data) @tool_register class cancel_reservation(ToolBase): """Playing a specific playlist by its name.""" reservation_id: int = Field(description="The unique identifier of the reservation to be canceled.") @classmethod def invoke(cls, input: Dict) -> str: reservation_id = input.get("reservation_id", None) missing = [] if not reservation_id: missing.append("reservation_id") if len(missing): value = ", ".join(missing) return f"Unable to cancel the reservation. The following required arguments are missing:{value}." instance = cls(**input) reservation_id = instance.reservation_id if reservation_id not in reservations: return f"There is no reservations with the id: {reservation_id}" reservations.pop(reservation_id) return f"The reservation {reservation_id} is cancled correctly" @tool_register class modify_reservation(ToolBase): """ Allows a user to modify an existing reservation by updating the check-in/check-out dates or changing the room type, subject to availability. One of these parameter is mandatory: new_room_type, new_reservation_start_date, new_reservation_end_date, guests. """ reservation_id: int = Field(description="The unique identifier of the reservation to be modified.") new_room_type: str | None = Field(default=list(json_data["room_types"].keys()), description=f"The type of new room to be modified, if {None} same room will be modified.") new_reservation_start_date: str = Field(default=None, description="New check out date in format DD/MM/YYYY") new_reservation_end_date: str = Field(default=None, description="New check out date in format DD/MM/YYYY") new_guests: int = Field(default=None, description="New number of guests for the reservation.") @classmethod def invoke(cls, input: Dict) -> str: input["new_room_type"] = input.get("new_room_type", None) reservation_id = input.get("reservation_id", None) missing = [] if not reservation_id: missing.append("reservation_id") instance = cls(**input) new_room_type = instance.new_room_type new_reservation_start_date = instance.new_reservation_start_date new_reservation_end_date = instance.new_reservation_end_date new_guests = instance.new_guests reservation_id = instance.reservation_id if len(missing): value = ", ".join(missing) return f"Unable to modify the reservation. The following required arguments are missing:{value}." if not (new_room_type or new_reservation_start_date or new_reservation_end_date or new_guests): return "Unable to modify the reservation. One of the following arguments must be passed: new_room_type, new_reservation_start_date, new_reservation_end_date, new_guests." if reservation_id not in reservations: return f"There is no reservations with the id: {reservation_id}" if new_room_type or new_guests: rooms = [room for room in json_data["accomodations"]["rooms"] if new_room_type in room["type"]] if len(rooms) == 0: return f"There is no room exists with room type {new_room_type}" rooms = [room for room in rooms if new_guests <= room["number_of_guests"]] if len(rooms) == 0: max_guests = json_data["room_types"][new_room_type]["number_of_guests"] return f"The number of guest is superior then the availibilty, maximum is {max_guests}" room = rooms[random.randint(0, len(rooms) - 1)] room_number = room["room_number"] else: room_number = reservations[reservation_id]["room_number"] reservations[reservation_id]["guests"] = new_guests if new_guests else reservations[reservation_id]["guests"] reservations[reservation_id]["reservation_start_date"] = new_reservation_start_date if new_reservation_start_date else reservations[reservation_id]["reservation_start_date"] reservations[reservation_id]["reservation_end_date"] = new_reservation_end_date if new_reservation_end_date else reservations[reservation_id]["reservation_end_date"] reservations[reservation_id]["room_type"] = new_room_type if new_room_type else reservations[reservation_id]["room_type"] reservations[reservation_id]["room_number"] = room_number tmp_data = reservations[reservation_id] return f"The reservation {reservation_id} is modified correctly: {json.dumps(tmp_data)}" @tool_register class get_reservation_detail(ToolBase): """Retrieve comprehensive details of a specific room reservation by providing its unique reservation ID.""" reservation_id: int = Field(description="Id of the reservation") @classmethod def invoke(cls, input: Dict) -> str: reservation_id = input.get("reservation_id", None) missing = [] if not reservation_id: missing.append("reservation_id") if len(missing): value = ", ".join(missing) return f"Unable to get the details. The following required arguments are missing:{value}." instance = cls(**input) reservation_id = instance.reservation_id if reservation_id not in reservations: return f"There is no reservations with the id: {reservation_id}" tmp_data = copy.deepcopy(reservations[reservation_id]) return json.dumps(tmp_data)