Spaces:
Sleeping
Sleeping
import gradio as gr | |
import requests | |
import uuid | |
from datetime import datetime | |
import os | |
# Get the API URL from environment variable or use default | |
API_URL = os.getenv('API_URL', 'http://localhost:8000') | |
API_TIMEOUT = 60 # seconds | |
def create_demo(): | |
# Initialize state | |
session_id = str(uuid.uuid4()) | |
messages = [] | |
suggested_questions = [] | |
def get_welcome_message(): | |
try: | |
welcome_response = requests.get(f"{API_URL}/welcome", timeout=API_TIMEOUT) | |
if welcome_response.status_code == 200: | |
welcome_data = welcome_response.json() | |
messages = [{"role": "assistant", "content": welcome_data["welcome_message"]}] | |
suggested_questions = welcome_data.get("suggested_questions", []) | |
return messages, gr.update(visible=True), suggested_questions | |
else: | |
fallback_message = "Welcome to the Longevity Assistant! How can I help you today?" | |
return [{"role": "assistant", "content": fallback_message}], gr.update(visible=False), [] | |
except Exception as e: | |
print(f"Error connecting to server: {str(e)}") | |
fallback_message = "Welcome to the Longevity Assistant! How can I help you today?" | |
return [{"role": "assistant", "content": fallback_message}], gr.update(visible=False), [] | |
def handle_chat(message, files, history): | |
# Add user message to history | |
history.append({"role": "user", "content": message}) | |
# If files were uploaded, add them to the message with distinct formatting | |
if files: | |
file_names = [f.name for f in files] | |
file_message = f"\n\nπ *Files attached:* {', '.join(file_names)} π" | |
history[-1]["content"] += file_message | |
try: | |
response = requests.post( | |
f"{API_URL}/chat", | |
json={"session_id": session_id, "message": message}, | |
timeout=API_TIMEOUT | |
) | |
if response.status_code == 200: | |
response_data = response.json() | |
history.append({"role": "assistant", "content": response_data["response"]}) | |
else: | |
error_message = "Sorry, I encountered an error while processing your request." | |
history.append({"role": "assistant", "content": error_message}) | |
except Exception as e: | |
error_message = "Sorry, I'm having trouble connecting to the server." | |
history.append({"role": "assistant", "content": error_message}) | |
return "", None, history | |
def handle_suggested_question(question, history): | |
return handle_chat(question, None, history) | |
def update_question_buttons(questions): | |
updates = [] | |
for i, button in enumerate(question_buttons): | |
if i < len(questions): | |
updates.append(gr.update(visible=True, value=questions[i])) | |
else: | |
updates.append(gr.update(visible=False)) | |
return updates | |
with gr.Blocks(css="#chatbot { height: 500px; } .upload-box { height: 40px; min-height: 40px; }") as demo: | |
gr.Markdown("# 𧬠Longevity Assistant") | |
chatbot = gr.Chatbot( | |
[], | |
elem_id="chatbot", | |
height=500, | |
type="messages" | |
) | |
with gr.Row(): | |
txt = gr.Textbox( | |
scale=4, | |
show_label=False, | |
placeholder="Enter text and press enter", | |
container=False | |
) | |
file_output = gr.File( | |
file_count="multiple", | |
label="", | |
scale=1, | |
container=False, | |
elem_classes="upload-box" | |
) | |
btn = gr.Button("Send", scale=1) | |
question_list = gr.State([]) | |
suggested_questions_container = gr.Column(visible=False) | |
with suggested_questions_container: | |
gr.Markdown("Try asking about:") | |
question_buttons = [] | |
# Create 4 rows with 2 buttons each (maximum 8 suggested questions) | |
for i in range(4): | |
with gr.Row(): | |
for j in range(2): | |
question_btn = gr.Button("", visible=False) | |
question_btn.click( | |
handle_suggested_question, | |
[question_btn, chatbot], | |
[txt, chatbot] | |
).then( | |
lambda: gr.update(visible=False), | |
None, | |
suggested_questions_container | |
) | |
question_buttons.append(question_btn) | |
# Update event handlers | |
txt.submit(handle_chat, [txt, file_output, chatbot], [txt, file_output, chatbot]) | |
btn.click(handle_chat, [txt, file_output, chatbot], [txt, file_output, chatbot]) | |
# Initialize welcome message and suggested questions | |
demo.load( | |
get_welcome_message, | |
outputs=[chatbot, suggested_questions_container, question_list] | |
).then( | |
update_question_buttons, | |
inputs=[question_list], | |
outputs=question_buttons | |
) | |
return demo | |
if __name__ == "__main__": | |
demo = create_demo() | |
demo.launch( | |
server_name="0.0.0.0", | |
server_port=7860, | |
share=True | |
) |