Spaces:
Running
Running
Create tools.py
Browse files
tools.py
ADDED
@@ -0,0 +1,341 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
1 |
+
import random
|
2 |
+
from abc import ABC, abstractmethod
|
3 |
+
import os
|
4 |
+
# from langchain.tools import tool
|
5 |
+
import json
|
6 |
+
from pydantic import BaseModel, Field
|
7 |
+
from openai import OpenAI
|
8 |
+
from typing import Dict, Optional, Union
|
9 |
+
import random
|
10 |
+
import copy
|
11 |
+
|
12 |
+
|
13 |
+
|
14 |
+
def read_json(data_path: str) -> tuple[list, dict]:
|
15 |
+
try:
|
16 |
+
with open(data_path, 'r', encoding="utf-8") as f:
|
17 |
+
data = [json.loads(line) for line in f.readlines()]
|
18 |
+
except:
|
19 |
+
with open(data_path, 'r', encoding="utf-8") as f:
|
20 |
+
data = json.loads(f.read())
|
21 |
+
return data
|
22 |
+
|
23 |
+
json_data = read_json("data/val_de_nuria.json")
|
24 |
+
reservations = {}
|
25 |
+
class ToolBase(BaseModel, ABC):
|
26 |
+
@abstractmethod
|
27 |
+
def invoke(cls, input: Dict):
|
28 |
+
pass
|
29 |
+
|
30 |
+
@classmethod
|
31 |
+
def to_openai_tool(cls):
|
32 |
+
"""
|
33 |
+
Extracts function metadata from a Pydantic class, including function name, parameters, and descriptions.
|
34 |
+
Formats it into a structure similar to OpenAI's function metadata.
|
35 |
+
"""
|
36 |
+
function_metadata = {
|
37 |
+
"type": "function",
|
38 |
+
"function": {
|
39 |
+
"name": cls.__name__, # Function name is same as the class name, in lowercase
|
40 |
+
"description": cls.__doc__.strip(),
|
41 |
+
"parameters": {
|
42 |
+
"type": "object",
|
43 |
+
"properties": {},
|
44 |
+
"required": [],
|
45 |
+
},
|
46 |
+
},
|
47 |
+
}
|
48 |
+
|
49 |
+
# Iterate over the fields to add them to the parameters
|
50 |
+
for field_name, field_info in cls.model_fields.items():
|
51 |
+
# Field properties
|
52 |
+
field_type = "string" # Default to string, will adjust if it's a different type
|
53 |
+
annotation = field_info.annotation.__args__[0] if getattr(field_info.annotation, "__origin__", None) is Union else field_info.annotation
|
54 |
+
|
55 |
+
if annotation == int:
|
56 |
+
field_type = "integer"
|
57 |
+
elif annotation == bool:
|
58 |
+
field_type = "boolean"
|
59 |
+
|
60 |
+
# Add the field's description and type to the properties
|
61 |
+
function_metadata["function"]["parameters"]["properties"][field_name] = {
|
62 |
+
"type": field_type,
|
63 |
+
"description": field_info.description,
|
64 |
+
}
|
65 |
+
|
66 |
+
# Determine if the field is required (not Optional or None)
|
67 |
+
if field_info.is_required():
|
68 |
+
function_metadata["function"]["parameters"]["required"].append(field_name)
|
69 |
+
|
70 |
+
# If there's an enum (like for `unit`), add it to the properties
|
71 |
+
if hasattr(field_info, 'default') and field_info.default is not None and isinstance(field_info.default, list):
|
72 |
+
function_metadata["function"]["parameters"]["properties"][field_name]["enum"] = field_info.default
|
73 |
+
|
74 |
+
return function_metadata
|
75 |
+
|
76 |
+
tools: Dict[str, ToolBase] = {}
|
77 |
+
oitools = []
|
78 |
+
|
79 |
+
def tool_register(cls: BaseModel):
|
80 |
+
oaitool = cls.to_openai_tool()
|
81 |
+
|
82 |
+
oitools.append(oaitool)
|
83 |
+
tools[oaitool["function"]["name"]] = cls
|
84 |
+
|
85 |
+
@tool_register
|
86 |
+
class get_hotel_information(ToolBase):
|
87 |
+
"""Retrieves basic information about the hotel, such as its name, address, contact details, and overall description."""
|
88 |
+
|
89 |
+
@classmethod
|
90 |
+
def invoke(cls, input: Dict) -> str:
|
91 |
+
return """### **Nou Vall de Núria – Brief Description**
|
92 |
+
|
93 |
+
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.
|
94 |
+
|
95 |
+
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.
|
96 |
+
|
97 |
+
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.
|
98 |
+
|
99 |
+
Whether for relaxation or adventure, **Nou Vall de Núria** promises a unique and memorable experience."""
|
100 |
+
|
101 |
+
@tool_register
|
102 |
+
class get_hotel_facilities(ToolBase):
|
103 |
+
"""Provides a list of available general facilities at the hotel, which could include amenities like a spa, pool, gym, conference rooms, etc."""
|
104 |
+
|
105 |
+
@classmethod
|
106 |
+
def invoke(cls, input: Dict) -> str:
|
107 |
+
return json_data["general_facilities"]
|
108 |
+
|
109 |
+
@tool_register
|
110 |
+
class get_restaurants_info(ToolBase):
|
111 |
+
"""Provides a list of available restaurants with their information."""
|
112 |
+
|
113 |
+
@classmethod
|
114 |
+
def invoke(cls, input: Dict) -> str:
|
115 |
+
"""
|
116 |
+
Play a playlist by its name, starting with the first or a random song.
|
117 |
+
"""
|
118 |
+
|
119 |
+
return json_data["restaurants"]
|
120 |
+
|
121 |
+
|
122 |
+
# @tool_register
|
123 |
+
# class get_restaurant_details(ToolBase):
|
124 |
+
# """Retrieves detailed information about a specific restaurant in the hotel, including its menu, ambiance, operating hours, and special features."""
|
125 |
+
|
126 |
+
# name: str = Field(default=[res["name"] for res in json_data["restaurants"]], description="Name of the resaturant")
|
127 |
+
|
128 |
+
# @classmethod
|
129 |
+
# def invoke(cls, input: Dict) -> str:
|
130 |
+
# """
|
131 |
+
# Play a playlist by its name, starting with the first or a random song.
|
132 |
+
# """
|
133 |
+
|
134 |
+
# instance = cls(**input)
|
135 |
+
# name = instance.name
|
136 |
+
|
137 |
+
# restaurante = [res for res in json_data["restaurants"] if res["name"] == name]
|
138 |
+
# if restaurante:
|
139 |
+
# return restaurante
|
140 |
+
# else:
|
141 |
+
# return f"We don't have any restaurante with the name: {name}"
|
142 |
+
|
143 |
+
|
144 |
+
@tool_register
|
145 |
+
class get_room_types(ToolBase):
|
146 |
+
"""
|
147 |
+
Returns a list of room types available at the hotel (e.g., single, double, suite, deluxe) along with brief descriptions of each type.
|
148 |
+
"""
|
149 |
+
@classmethod
|
150 |
+
def invoke(cls, input: Dict) -> str:
|
151 |
+
return json_data["room_types"]
|
152 |
+
|
153 |
+
|
154 |
+
@tool_register
|
155 |
+
class check_room_availability(ToolBase):
|
156 |
+
"""
|
157 |
+
Checks if a specified room type is available between the provided check-in and check-out dates for a given number of guests.
|
158 |
+
"""
|
159 |
+
room_type: str = Field(default=list(json_data["room_types"].keys()), description="The type of room the user is interested in")
|
160 |
+
check_in_date: str = Field(description="The starting date of the reservation (e.g., \"2025-04-01\")")
|
161 |
+
check_out_date: str = Field(description="The ending date of the reservation (e.g., \"2025-04-05\").")
|
162 |
+
guests: int = Field(description="The number of guests that will occupy the room.")
|
163 |
+
|
164 |
+
@classmethod
|
165 |
+
def invoke(cls, input: Dict) -> str:
|
166 |
+
|
167 |
+
instance = cls(**input)
|
168 |
+
room_type = instance.room_type
|
169 |
+
check_in_date = instance.check_in_date
|
170 |
+
check_in_date = instance.check_in_date
|
171 |
+
guests = instance.guests
|
172 |
+
rooms = [room for room in json_data["restaurants"] if room_type in room["type"]]
|
173 |
+
if len(rooms) == 0:
|
174 |
+
return f"There is no room exists with room type {room_type}"
|
175 |
+
|
176 |
+
rooms2 = [room for room in rooms if guests >= room["number_of_guests"]]
|
177 |
+
if len(rooms2) == 0:
|
178 |
+
max_guests = json_data["room_types"][room_type]["number_of_guests"]
|
179 |
+
return f"The number of guest is superior then the availibilty, maximum is {max_guests}"
|
180 |
+
|
181 |
+
|
182 |
+
@tool_register
|
183 |
+
class make_reservation(ToolBase):
|
184 |
+
"""
|
185 |
+
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.
|
186 |
+
"""
|
187 |
+
|
188 |
+
room_type: str = Field(default=list(json_data["room_types"].keys()), description="The type of room being reserved.")
|
189 |
+
check_in_date: str = Field(description="The starting date of the reservation (e.g., \"2025-04-01\")")
|
190 |
+
check_out_date: str = Field(description="The ending date of the reservation (e.g., \"2025-04-05\").")
|
191 |
+
guests: int = Field(description="The number of guests for the reservation.")
|
192 |
+
user_id: int = Field(description="The identifier for the user making the reservation.")
|
193 |
+
|
194 |
+
@classmethod
|
195 |
+
def invoke(cls, input: Dict) -> str:
|
196 |
+
|
197 |
+
instance = cls(**input)
|
198 |
+
room_type = instance.room_type
|
199 |
+
check_in_date = instance.check_in_date
|
200 |
+
check_in_date = instance.check_in_date
|
201 |
+
guests = instance.guests
|
202 |
+
user_id = instance.user_id
|
203 |
+
|
204 |
+
|
205 |
+
rooms = [room for room in json_data["restaurants"] if room_type in room["type"]]
|
206 |
+
if len(rooms) == 0:
|
207 |
+
return f"There is no room exists with room type {room_type}"
|
208 |
+
|
209 |
+
rooms2 = [room for room in rooms if guests >= room["number_of_guests"]]
|
210 |
+
if len(rooms2) == 0:
|
211 |
+
max_guests = json_data["room_types"][room_type]["number_of_guests"]
|
212 |
+
return f"The number of guest is superior then the availibilty, maximum is {max_guests}"
|
213 |
+
|
214 |
+
room = rooms2[random.randint(0, len(rooms2))]
|
215 |
+
|
216 |
+
rand = int(random.randint(0,10000000))
|
217 |
+
while rand in reservations:
|
218 |
+
rand = int(random.randint(0,10000000))
|
219 |
+
|
220 |
+
tmp_data = {
|
221 |
+
"status": "Reserved",
|
222 |
+
"room_number": room["room_number"],
|
223 |
+
"room_type": room_type,
|
224 |
+
"check_in_date": check_in_date,
|
225 |
+
"check_in_date": check_in_date,
|
226 |
+
"guests": guests,
|
227 |
+
"reservation_id": rand,
|
228 |
+
"user_id": user_id,
|
229 |
+
}
|
230 |
+
|
231 |
+
reservations[rand] = tmp_data
|
232 |
+
|
233 |
+
return json.dumps(tmp_data)
|
234 |
+
|
235 |
+
@tool_register
|
236 |
+
class cancel_reservation(ToolBase):
|
237 |
+
"""Playing a specific playlist by its name."""
|
238 |
+
|
239 |
+
user_id: int = Field(description="The identifier for the user requesting the cancellation.")
|
240 |
+
reservation_id: int = Field(description="The unique identifier of the reservation to be canceled.")
|
241 |
+
|
242 |
+
@classmethod
|
243 |
+
def invoke(cls, input: Dict) -> str:
|
244 |
+
"""
|
245 |
+
Play a playlist by its name, starting with the first or a random song.
|
246 |
+
"""
|
247 |
+
|
248 |
+
instance = cls(**input)
|
249 |
+
reservation_id = instance.reservation_id
|
250 |
+
user_id = instance.user_id
|
251 |
+
|
252 |
+
if reservation_id not in reservations:
|
253 |
+
return f"The is no reservations with the id: {reservation_id}"
|
254 |
+
|
255 |
+
if reservations["reservation_id"]["user_id"] != user_id:
|
256 |
+
return "The user id is wrong, please provide same user id that was used make to reservation."
|
257 |
+
|
258 |
+
reservations.pop(reservation_id)
|
259 |
+
return f"The reservation {reservation_id} is cancled correctly"
|
260 |
+
|
261 |
+
@tool_register
|
262 |
+
class modify_reservation(ToolBase):
|
263 |
+
"""
|
264 |
+
Allows a user to modify an existing reservation by updating the check-in/check-out dates or changing the room type, subject to availability.
|
265 |
+
"""
|
266 |
+
|
267 |
+
|
268 |
+
new_room_type: str = Field(default=list(json_data["room_types"].keys()) + [None], description=f"The type of new room to be modified, if {None} same room will be modified.")
|
269 |
+
new_check_in_date: str = Field(default=None, description="New check out date in format DD/MM/YYYY")
|
270 |
+
new_check_out_date: str = Field(default=None, description="New check out date in format DD/MM/YYYY")
|
271 |
+
guests: int = Field(default=None, description="New number of guests for the reservation.")
|
272 |
+
user_id: int = Field(description="The identifier for the user requesting the modification.")
|
273 |
+
reservation_id: int = Field(description="The unique identifier of the reservation to be modified.")
|
274 |
+
|
275 |
+
@classmethod
|
276 |
+
def invoke(cls, input: Dict) -> str:
|
277 |
+
|
278 |
+
instance = cls(**input)
|
279 |
+
new_room_type = instance.new_room_type
|
280 |
+
new_check_in_date = instance.new_check_in_date
|
281 |
+
new_check_out_date = instance.new_check_out_date
|
282 |
+
guests = instance.guests
|
283 |
+
user_id = instance.user_id
|
284 |
+
reservation_id = instance.reservation_id
|
285 |
+
|
286 |
+
if reservation_id not in reservations:
|
287 |
+
return f"The is no reservations with the id: {reservation_id}"
|
288 |
+
|
289 |
+
if reservations["reservation_id"]["user_id"] != user_id:
|
290 |
+
return "The user id is wrong, please provide same user id that was used make to reservation."
|
291 |
+
|
292 |
+
if new_room_type or guests:
|
293 |
+
rooms = [room for room in json_data["restaurants"] if new_room_type in room["type"]]
|
294 |
+
if len(rooms) == 0:
|
295 |
+
return f"There is no room exists with room type {new_room_type}"
|
296 |
+
|
297 |
+
rooms = [room for room in rooms if guests >= room["number_of_guests"]]
|
298 |
+
if len(rooms) == 0:
|
299 |
+
max_guests = json_data["room_types"][new_room_type]["number_of_guests"]
|
300 |
+
return f"The number of guest is superior then the availibilty, maximum is {max_guests}"
|
301 |
+
|
302 |
+
room = rooms[random.randint(0, len(rooms))]
|
303 |
+
room_number = room["room_number"]
|
304 |
+
else:
|
305 |
+
room_number = reservations["reservation_id"]["room_number"]
|
306 |
+
|
307 |
+
|
308 |
+
reservations["reservation_id"]["guests"] = guests if guests else reservations["reservation_id"]["guests"]
|
309 |
+
reservations["reservation_id"]["check_in_date"] = new_check_in_date if new_check_in_date else reservations["reservation_id"]["check_in_date"]
|
310 |
+
reservations["reservation_id"]["check_out_date"] = new_check_out_date if new_check_out_date else reservations["reservation_id"]["check_out_date"]
|
311 |
+
reservations["reservation_id"]["room_type"] = new_room_type if new_room_type else reservations["reservation_id"]["room_type"]
|
312 |
+
reservations["reservation_id"]["room_number"] = room_number
|
313 |
+
|
314 |
+
tmp_data = copy.deepcopy(reservations["reservation_id"])
|
315 |
+
tmp_data.pop("user_id")
|
316 |
+
|
317 |
+
return f"The reservation {reservation_id} is modified correctly: {json.dumps(tmp_data)}"
|
318 |
+
|
319 |
+
@tool_register
|
320 |
+
class get_reservation_details(ToolBase):
|
321 |
+
"""Playing a specific playlist by its name."""
|
322 |
+
|
323 |
+
user_id: int = Field(description="Id of user, could be passport or national Identity number")
|
324 |
+
reservation_id: int = Field(description="Id of the reservation")
|
325 |
+
|
326 |
+
@classmethod
|
327 |
+
def invoke(cls, input: Dict) -> str:
|
328 |
+
|
329 |
+
instance = cls(**input)
|
330 |
+
user_id = instance.user_id
|
331 |
+
reservation_id = instance.reservation_id
|
332 |
+
|
333 |
+
if reservation_id not in reservations:
|
334 |
+
return f"The is no reservations with the id: {reservation_id}"
|
335 |
+
|
336 |
+
if reservations["reservation_id"]["user_id"] != user_id:
|
337 |
+
return "The user id is wrong, please provide same user id that was used make to reservation."
|
338 |
+
|
339 |
+
tmp_data = copy.deepcopy(reservations["reservation_id"])
|
340 |
+
tmp_data.pop("user_id")
|
341 |
+
return json.dumps(tmp_data)
|