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.vectorstores import FAISS from langchain.embeddings import HuggingFaceEmbeddings class VectorStore: def __init__(self, embeddings_model, vectorstore): embeddings = HuggingFaceEmbeddings(model_name=embeddings_model, model_kwargs={'device': 'cpu'}) 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_documents(ToolBase): """ Retrieves general information about a region, its cities, activities, tourism, or surrounding areas based on query. """ query: str = Field(description="An enhanced user query optimized for retrieving 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 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): """ Returns a list information about rooms available at the hotel along with brief descriptions of each 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 the user is interested in") 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: 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: 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. """ 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") guests: int = Field(default=None, description="New number of guests for the reservation.") reservation_id: int = Field(description="The unique identifier of the reservation to be modified.") @classmethod def invoke(cls, input: Dict) -> str: 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 guests = instance.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 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, guests." if reservation_id not in reservations: return f"There is no reservations with the id: {reservation_id}" if new_room_type or guests: rooms = [room for room in json_data["room_types"] 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 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"] = guests if 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 reservation_details(ToolBase): """Playing a specific playlist by its name.""" 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)