Initial commit (copied from hotel_tools space)

.gitignore

@@ -0,0 +1,7 @@
+venv/
+.env
+__pycache__
+.qodo
+data/
+.DS_Store
+

app.py

@@ -0,0 +1,137 @@
+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
+
+load_dotenv(".env")
+HF_TOKEN = os.environ.get("HF_TOKEN")  
+BASE_URL = os.environ.get("BASE_URL")  
+
+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.  
+
+### **Response Guidelines:**  
+- **Accuracy & Completeness:** Never provide information that is not explicitly available. If a required parameter (e.g., check-in date, check-out date, number of guests, or room type) is missing, ask the user to provide it before proceeding.  
+- **No Hallucinations:** Do not assume details. If you are unsure about something, ask the user for clarification.  
+- **Consistency:** Always respond in the same language as the user’s query. 
+
+### **Booking Requirements:**  
+You **can only process hotel room reservations**.
+If a user requests a **restaurant, flight, or any other type of reservation**, immediately inform them that you **cannot process those reservations**.
+Do **not** ask for further details about restaurant or flight bookings. Simply clarify that you **only handle hotel room reservations**.  
+
+Today’s date is **{date}**."""
+
+
+# print(json.dumps(oitools, indent=2))
+client = OpenAI(
+    base_url=f"{BASE_URL}/v1",  
+    api_key=HF_TOKEN
+)
+
+def today_date():
+    return datetime.today().strftime('%A, %B %d, %Y, %I:%M %p')
+
+
+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:
+        if isinstance(msg, dict):  
+            msg = ChatMessage(**msg)
+        if msg.role == "assistant" and hasattr(msg, "metadata") and msg.metadata:  
+            tools_calls = json.loads(msg.metadata.get("title", "[]")) 
+            # for tool_calls in tools_calls:
+            #     tool_calls["function"]["arguments"] = json.loads(tool_calls["function"]["arguments"])
+            messages.append({"role": "assistant", "tool_calls": tools_calls, "content": ""})
+            messages.append({"role": "tool", "content": msg.content})
+        else:
+            messages.append({"role": msg.role, "content": msg.content})
+    
+    request_params = {
+        "model": model,
+        "messages": messages,
+        "stream": True,
+        "max_tokens": 1000,
+        "temperature": 0.2,
+        "frequency_penalty": 1,
+        "extra_body": {"repetition_penalty": 1.1},
+    }
+    if tools:
+        request_params.update({"tool_choice": "auto", "tools": tools})
+    
+    return client.chat.completions.create(**request_params)  
+
+def llm_in_loop(history, system_prompt, recursive):  
+    try:   
+        models = client.models.list()
+        model = models.data[0].id if models.data else "gpt-3.5-turbo"  
+    except Exception as err:
+        gr.Warning("The model is initializing. Please wait; this may take 5 to 10 minutes ⏳.", duration=20)
+        raise err
+    
+    arguments = ""
+    name = ""
+    chat_completion = completion(history=history, tools=oitools, model=model, system_prompt=system_prompt)  
+    appended = False
+    # if chat_completion.choices and chat_completion.choices[0].message.tool_calls:
+    #     call = chat_completion.choices[0].message.tool_calls[0]
+    #     if hasattr(call.function, "name") and call.function.name:
+    #         name = call.function.name
+    #     if hasattr(call.function, "arguments") and call.function.arguments:
+    #         arguments += call.function.arguments
+    # elif chat_completion.choices[0].message.content:
+    #     if not appended:
+    #         history.append(ChatMessage(role="assistant", content=""))
+    #         appended = True
+    #     history[-1].content += chat_completion.choices[0].message.content
+    #     yield history[recursive:]
+    for chunk in chat_completion:
+        if chunk.choices and chunk.choices[0].delta.tool_calls:
+            call = chunk.choices[0].delta.tool_calls[0]
+            if hasattr(call.function, "name") and call.function.name:
+                name = call.function.name
+            if hasattr(call.function, "arguments") and call.function.arguments:
+                arguments += call.function.arguments
+        elif chunk.choices[0].delta.content:
+            if not appended:
+                history.append(ChatMessage(role="assistant", content=""))
+                appended = True
+            history[-1].content += chunk.choices[0].delta.content
+            yield history[recursive:]
+    
+    arguments = clean_json_string(arguments) if arguments else "{}"
+    print(name, arguments)
+    arguments = json.loads(arguments)
+    print(name, arguments)
+    print("====================")
+    if appended:
+        recursive -= 1
+    if name:
+        result = f"💥 Error using tool {name}, tool doesn't exist" if name not in tools else str(tools[name].invoke(input=arguments))
+        result = json.dumps({name: result}, ensure_ascii=False)
+        # msg = ChatMessage(
+        #             role="assistant",
+        #             content="",
+        #             metadata= {"title": f"🛠️ Using tool '{name}', arguments: {json.dumps(json_arguments, ensure_ascii=False)}"},
+        #             options=[{"label":"tool_calls", "value": json.dumps([{"id": "call_FthC9qRpsL5kBpwwyw6c7j4k","function": {"arguments": arguments,"name": name},"type": "function"}])}]
+        #         )
+        history.append(ChatMessage(role="assistant", content=result, metadata={"title": json.dumps([{"id": "call_id", "function": {"arguments": json.dumps(arguments, ensure_ascii=False), "name": name}, "type": "function"}], ensure_ascii=False)}))
+        yield history[recursive:]
+        yield from llm_in_loop(history, system_prompt, recursive - 1)
+
+def respond(message, history, additional_inputs):  
+    history.append(ChatMessage(role="user", content=message))
+    yield from llm_in_loop(history, additional_inputs, -1)
+
+if __name__ == "__main__":
+    system_prompt = gr.Textbox(label="System prompt", value=SYSTEM_PROMPT_TEMPLATE, lines=3)  
+    demo = gr.ChatInterface(respond, type="messages", additional_inputs=[system_prompt])
+    demo.launch()

managed_context/metadata.json

@@ -0,0 +1 @@
+{"current_schema_version":"0.0.1"}

requirements.txt

@@ -0,0 +1,8 @@
+gradio==5.23.0
+openai==1.68.2
+python-dotenv==1.1.0
+langchain-community==0.3.20
+langchain-core==0.3.48
+faiss-cpu==1.10.0
+faiss-gpu==1.7.2
+sentence-transformers==3.4.1

test_suite_analysis/metadata.json

@@ -0,0 +1 @@
+{"current_schema_version":"0.0.1"}

tools.py

@@ -0,0 +1,458 @@
+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_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 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)