Adapted from Hotel Tools

.gitignore

@@ -4,4 +4,3 @@ __pycache__
 .qodo
 data/
 .DS_Store
-

app.py

@@ -1,17 +1,30 @@
+from dotenv import load_dotenv
+
+
 import gradio as gr
 from gradio import ChatMessage
 
 import json
 from openai import OpenAI
-from tools import tools, oitools
-from dotenv import load_dotenv
 from datetime import datetime
 import os
 import re
+import logging
+
+logging.basicConfig(level=logging.INFO, format='[%(asctime)s][%(levelname)s] - %(message)s')
+# logging.getLogger().setLevel(logging.INFO)
+
 
 load_dotenv(".env")
 HF_TOKEN = os.environ.get("HF_TOKEN")  
-BASE_URL = os.environ.get("BASE_URL")  
+BASE_URL = os.environ.get("BASE_URL")
+EMBEDDINGS = os.environ.get("EMBEDDINGS")
+
+print("HF_TOKEN: ", HF_TOKEN)
+print("BASE_URL: ", BASE_URL)
+
+
+from tools import tools, oitools
 
 SYSTEM_PROMPT_TEMPLATE = """You are an AI assistant designed to assist users with a **hotel booking and information system**. Your primary role is to provide detailed and accurate information about the hotel, including available accommodations, facilities, dining options, and reservation services. You can assist with **hotel room bookings**, modify or cancel reservations, and answer general inquiries about the hotel.  
 
@@ -28,11 +41,11 @@ Do **not** ask for further details about restaurant or flight bookings. Simply c
 Today’s date is **{date}**."""
 
 
-# print(json.dumps(oitools, indent=2))
 client = OpenAI(
     base_url=f"{BASE_URL}/v1",  
     api_key=HF_TOKEN
 )
+logging.info(f"Client initialized: {client}")
 
 def today_date():
     return datetime.today().strftime('%A, %B %d, %Y, %I:%M %p')
@@ -41,6 +54,7 @@ def today_date():
 def clean_json_string(json_str):
     return re.sub(r'[ ,}\s]+$', '', json_str) + '}'
 
+
 def completion(history, model, system_prompt: str, tools=None):
     messages = [{"role": "system", "content": system_prompt.format(date=today_date())}]
     for msg in history:

src/vectorstore.py

@@ -0,0 +1,40 @@
+from langchain_community.vectorstores import FAISS
+from langchain_community.embeddings import HuggingFaceEmbeddings
+from huggingface_hub import snapshot_download
+import logging
+
+
+class VectorStore:
+    def __init__(self, embeddings_model, vs_local_path=None, vs_hf_path=None, number_of_contexts=2):
+        
+        self.number_of_contexts = number_of_contexts
+
+        logging.info("Loading vectorstore...")        
+        
+        embeddings = HuggingFaceEmbeddings(model_name=embeddings_model)
+        logging.info(f"Loaded embeddings model: {embeddings_model}")
+
+
+        if vs_hf_path:
+            hf_vectorstore = snapshot_download(repo_id=vs_hf_path)
+            self.vectore_store = FAISS.load_local(hf_vectorstore, embeddings, allow_dangerous_deserialization=True)
+            logging.info(f"Loaded vectorstore from {vs_hf_path}")
+        else:
+            self.vectore_store = FAISS.load_local(vs_local_path, embeddings, allow_dangerous_deserialization=True)
+            logging.info(f"Loaded vectorstore from {vs_local_path}")
+
+    
+    def get_context(self, instruction, number_of_contexts=2):
+
+        logging.info(f"Getting context for instruction: {instruction}")
+        documentos = self.vectore_store.similarity_search_with_score(instruction, k=self.number_of_contexts)
+        return self._beautiful_context(documentos)
+    
+    def _beautiful_context(self, docs):
+        context = ""
+        for doc in docs:
+            context += doc[0].page_content + "\n"
+
+        print("Context: ", context)
+        return context
+

tools.py

@@ -1,44 +1,14 @@
-import random
+
 from abc import ABC, abstractmethod
-from typing import get_origin, get_args
-import os
-# from langchain.tools import tool
-import json
+from typing import Dict, Union, get_origin, get_args
 from pydantic import BaseModel, Field
-from typing import Dict,  Union
-import random
-import copy
 from types import UnionType
+import os
+import logging
 
-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
+from src.vectorstore import VectorStore
+# from langchain.tools import tool
 
-json_data = read_json("data/val_de_nuria.json")
-reservations = {}
 
 class ToolBase(BaseModel, ABC):
     @abstractmethod
@@ -109,7 +79,14 @@ class ToolBase(BaseModel, ABC):
 
 tools: Dict[str, ToolBase] = {}
 oitools = []
-vector_store = VectorStore(embeddings_model="BAAI/bge-m3", vectorstore="data/vs")
+
+
+vector_store = VectorStore(
+    # embeddings_model="BAAI/bge-m3", 
+    embeddings_model=os.environ.get("EMBEDDINGS_MODEL"),
+    vs_local_path=os.environ.get("VS_LOCAL_PATH"), 
+    vs_hf_path=os.environ.get("VS_HF_PATH"))
+
 
 def tool_register(cls: BaseModel):
     oaitool = cls.to_openai_tool()
@@ -117,28 +94,24 @@ def tool_register(cls: BaseModel):
     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.
     """
+    
+    logging.info("@tool_register: get_documents()")  
+    
     query: str = Field(description="An enhanced user query optimized for retrieving information")
+    logging.info(f"query: {query}")
 
     @classmethod
     def invoke(cls, input: Dict) -> str:
+
+        logging.info(f"get_documents.invoke() input: {input}")
+        # Check if the input is a dictionary
+
         query = input.get("query", None)
         if not query:
             return "Missing required argument: query."
@@ -146,313 +119,3 @@ class get_documents(ToolBase):
         # 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: 
-        
-        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)