Spaces:
Running
Running
File size: 7,234 Bytes
8575cb5 a255fea 8575cb5 a255fea d240252 a255fea 3721bf6 a255fea 5f9dbfa a255fea fa7fdf2 03a3aa7 fa7fdf2 313bd80 fa7fdf2 8575cb5 bd634a3 a255fea 5f9dbfa 8575cb5 d240252 3721bf6 3aac3b1 3721bf6 d240252 a255fea 5f9dbfa a255fea 5f9dbfa bd634a3 a255fea 5f9dbfa d240252 bd634a3 d240252 5f9dbfa d240252 5f9dbfa d240252 2e8758e e223bec 5f9dbfa 8575cb5 5f9dbfa a255fea 5f9dbfa a255fea 5f9dbfa a255fea 5f9dbfa a255fea d240252 3721bf6 bd634a3 5f9dbfa a255fea 5f9dbfa 76d9486 5f9dbfa 8575cb5 5f9dbfa 8575cb5 5f9dbfa 80b38e7 5f9dbfa |
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 |
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.
- **Relevance & Clarity:** Ensure responses are clear, concise, and directly relevant to the query.
### **Booking Requirements:**
You **can only process hotel room reservations**. Before proceeding with a reservation, always ask for the following details:
1. **Check-in date**
2. **Check-out date**
3. **Number of guests**
4. **Room type**
### **Restrictions on Other 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**.
### **Handling Additional Requests:**
- If the user asks for information about the hotel’s **history, the surrounding region, cities, activities, tourism, or nearby attractions**, use the function `get_documents` to fetch reliable data instead of making assumptions.
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})
for msg in messages:
print(msg)
print("")
print("")
request_params = {
"model": model,
"messages": messages,
"stream": True,
"max_tokens": 1000,
"temperature": 0.15,
"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)
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()
|