File size: 10,520 Bytes
4357218 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 |
from dotenv import load_dotenv
from openai import OpenAI
import json
import os
import requests
from pathlib import Path
from pypdf import PdfReader
from pydantic import BaseModel
import gradio as gr
load_dotenv(override=True)
pushover_token = os.getenv("PUSHOVER_TOKEN")
pushover_user = os.getenv("PUSHOVER_USER")
pushover_base_url = "https://api.pushover.net/1/messages.json"
gemini_api_key = os.getenv("GEMINI_API_KEY")
GEMINI_BASE_URL = "https://generativelanguage.googleapis.com/v1beta/openai/"
class Evaluation(BaseModel):
is_acceptable: bool
feedback: str
def push(text):
requests.post(
pushover_base_url,
data = {
"token": pushover_token,
"user": pushover_user,
"message": text
}
)
def record_booking_table(email, name = "name not provided", date = "date not provided", time = "time not provided"):
push(f"{name} has booked tables on {date} at {time}. You can connect him via {email}")
return {"recorded": "ok"}
def record_feedback(feedback):
push(f"Somebody has left feedback: {feedback}")
return {"recorded": "ok"}
record_booking_table_json = {
"name" : "record_booking_table",
"description" : "Use this tool to record, when a user wants to book a table and provided his email.",
"parameters": {
"type": "object",
"properties":{
"email":{
"type": "string",
"description": "The email address of a user"
},
"name":{
"type": "string",
"description": "The user's name, if they provided it"
},
"date":{
"type": "string",
"description": "Date, when user wants to book a table like (xxth of month), if they provided it."
},
"time":{
"type": "string",
"description": "Time, when user wants to book a table like (hh:mm), if they provided it."
}
},
"required": ["email"],
"additionalProperties": False
}
}
record_feedback_json = {
"name": "record_feedback",
"description": "Always use this tool to record structured feedback about our restaurant. If feedback isn`t related to restaurant, don`t use this tool.",
"parameters":{
"type": "object",
"properties":{
"feedback": {
"type": "string",
"description": "The feedback that was given."
}
},
"required": ["feedback"],
"additionalProperties": False
}
}
tools = [{"type": "function", "function": record_booking_table_json},
{"type": "function", "function": record_feedback_json}]
class RestaurantBot:
def __init__(self):
self.gemini = OpenAI(base_url = GEMINI_BASE_URL, api_key=gemini_api_key)
self.name = "Night City"
pdf_path = Path(__file__).resolve().parent / "menu_restorana.pdf"
reader = PdfReader(str(pdf_path))
self.menu = ""
for page in reader.pages:
text = page.extract_text()
if text:
self.menu += text
def handle_tool_call(self, tool_calls):
results =[]
for tool_call in tool_calls:
tool_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
tool = globals().get(tool_name)
result = tool(**arguments) if tool else {}
results.append({"role": "tool", "content": json.dumps(result), "tool_call_id": tool_call.id})
return results
def system_prompt(self):
system_instructions = f"""
You are Alexander, the professional and welcoming head hostess of the '{self.name}' restaurant.
Your primary goal is to assist guests with the menu, answer questions about dishes, and facilitate table reservations.
### OPERATIONAL DETAILS:
* **Working Hours**: We are open from 10:00 to 22:00 every day.
* **Contact Numbers**: (097) 454 6555 or (063) 584 65 55.
* **Official Website**: night-city.com.ua[cite: 7, 11].
### GUIDELINES:
1. **Persona**: Be polite, sophisticated, and helpful. Treat every user as a valued guest entering our establishment.
2. **Menu Knowledge**: Use the provided menu context to describe dishes, ingredients, and prices.
3. **Upselling**: When a guest shows interest in meat or fish dishes, suggest a matching wine from our Georgian or French collections.
4. **Reservations**:
- Actively encourage guests to book a table if they seem interested in visiting.
- You MUST obtain an **email address** to complete a booking.
- Try to collect the guest's name, preferred date, and time.
- Once you have the email, use the `record_booking_table` tool immediately.
5. **Feedback & Edge Cases**:
- If a guest wants to leave a review, complaint, or suggestion, use the `record_feedback` tool.
- If you are asked a question you cannot answer based on the menu, or if the request is unrelated to the restaurant, politely inform the guest that you will pass this to the manager and use the `record_feedback` tool to log it.
### RESTAURANT MENU:
{self.menu}
Act as Alexander. Always stay in character and respond in the language used by the guest.
"""
return system_instructions
def evaluator_system_prompt(self):
# Follow the evaluation pattern from 3_lab3.ipynb, adapted to the restaurant persona
system = (
"You are an evaluator that decides whether a response to a restaurant guest is acceptable.\n"
"You are provided with a conversation between a Guest and Alexander (the restaurant host).\n"
"Your task is to decide whether Alexander's latest response is acceptable quality.\n"
"Alexander must be polite, professional, knowledgeable about the menu, and helpful with reservations.\n"
"Alexander has been given detailed instructions and the full restaurant menu as context:\n\n"
f"{self.system_prompt()}\n\n"
"With this context, please evaluate the latest response, replying with whether the response is acceptable and your feedback."
)
return system
def evaluator_user_prompt(self, reply, message, history):
user_prompt = f"Here's the conversation between the Guest and Alexander: \n\n{history}\n\n"
user_prompt += f"Here's the latest message from the Guest: \n\n{message}\n\n"
user_prompt += f"Here's the latest response from Alexander: \n\n{reply}\n\n"
user_prompt += "Please evaluate the response, replying with whether it is acceptable and your feedback."
return user_prompt
def evaluate(self, reply, message, history) -> Evaluation:
messages = [
{"role": "system", "content": self.evaluator_system_prompt()},
{"role": "user", "content": self.evaluator_user_prompt(reply, message, history)},
]
response = self.gemini.beta.chat.completions.parse(
model="gemini-2.5-flash", messages=messages, response_format=Evaluation
)
return response.choices[0].message.parsed
def rerun(self, reply, message, history, feedback):
# Same rerun pattern as in 3_lab3.ipynb: augment system prompt with rejection info
updated_system_prompt = self.system_prompt() + "\n\n## Previous answer rejected\n"
updated_system_prompt += "You just tried to reply, but the quality control rejected your reply.\n"
updated_system_prompt += f"## Your attempted answer:\n{reply}\n\n"
updated_system_prompt += f"## Reason for rejection:\n{feedback}\n\n"
messages = [{"role": "system", "content": updated_system_prompt}] + history + [
{"role": "user", "content": message}
]
response = self.gemini.chat.completions.create(
model="gemini-2.5-flash", messages=messages, tools=tools
)
return response.choices[0].message.content
def chat(self, message, history):
# Clean history like in the lab note so providers don't choke on extra keys
history = [{"role": h["role"], "content": h["content"]} for h in history]
messages = [{"role": "system", "content": self.system_prompt()}] + history + [
{"role": "user", "content": message}
]
done = False
response = None
while not done:
response = self.gemini.chat.completions.create(
model="gemini-2.5-flash", messages=messages, tools=tools
)
if response.choices[0].finish_reason == "tool_calls":
msg_obj = response.choices[0].message
tool_calls = msg_obj.tool_calls
results = self.handle_tool_call(tool_calls)
# Store a plain-dict version of the assistant message
if hasattr(msg_obj, "model_dump"):
messages.append(msg_obj.model_dump())
elif hasattr(msg_obj, "to_dict"):
messages.append(msg_obj.to_dict())
else:
messages.append(
{
"role": "assistant",
"content": getattr(msg_obj, "content", None),
"tool_calls": tool_calls,
}
)
messages.extend(results)
else:
done = True
reply = response.choices[0].message.content
# Run evaluation just like in 3_lab3.ipynb
evaluation = self.evaluate(reply, message, history)
if evaluation.is_acceptable:
print("Passed evaluation - returning reply")
return reply
else:
print("Failed evaluation - retrying")
print(evaluation.feedback)
return self.rerun(reply, message, history, evaluation.feedback)
if __name__ == "__main__":
restaurant_bot = RestaurantBot()
gr.ChatInterface(restaurant_bot.chat, type="messages").launch() |