forked from elzanou/mAICookbook
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathapp.py
131 lines (108 loc) · 4.55 KB
/
app.py
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
import datetime
import requests
import streamlit as st
# FastAPI server details
API_URL = "http://127.0.0.1:8000/chat"
START_CONVERSATION_URL = "http://127.0.0.1:8000/start_conversation"
ALL_CONVERSATIONS_URL = "http://127.0.0.1:8000/all_conversations"
HISTORY_URL = "http://127.0.0.1:8000/history"
STATUS_URL = "http://127.0.0.1:8000/status"
# Retry parameters
RETRY_TIMEOUT = 10 # Total timeout in seconds
RETRY_ATTEMPTS = 5 # Interval between retries in seconds
# Generate a localized welcome message
def generate_welcome_message():
dt = datetime.datetime.now()
return f"{'Καλημέρα' if dt.hour < 12 else 'Καλησπέρα'}. Τι καλό θα ήθελες να φτιάξουμε;"
# Function to start a new conversation
def start_new_conversation():
try:
response = requests.post(START_CONVERSATION_URL)
response.raise_for_status()
return response.json().get("conversation_id")
except requests.exceptions.RequestException:
st.error("Failed to start a new conversation. Please try again.")
return None
# Function to fetch all conversations
def fetch_all_conversations():
try:
response = requests.get(ALL_CONVERSATIONS_URL)
response.raise_for_status()
return response.json().get("conversations", [])
except requests.exceptions.RequestException:
st.error("Failed to fetch conversations. Please try again.")
return []
# Function to fetch a conversation history
def fetch_conversation_history(conversation_id):
try:
response = requests.get(f"{HISTORY_URL}/{conversation_id}")
response.raise_for_status()
return response.json().get("history", [])
except requests.exceptions.RequestException:
st.error("Failed to fetch conversation history. Please try again.")
return []
# Function to send a user message to the server
def query_fastapi(user_message, conversation_id):
try:
payload = {
"conversation_id": conversation_id,
"user_message": user_message,
"messages": st.session_state.messages
}
response = requests.post(API_URL, json=payload)
response.raise_for_status()
return response.json().get("response", "No response from server.")
except requests.exceptions.RequestException:
return "I'm having trouble connecting to the server. Please try again later."
# Initialize session state
if "conversation_id" not in st.session_state:
# Start a new conversation when the app loads
new_conversation_id = start_new_conversation()
if new_conversation_id:
st.session_state.conversation_id = new_conversation_id
st.session_state.messages = [{"role": "assistant", "content": generate_welcome_message()}]
if "messages" not in st.session_state:
st.session_state.messages = [{"role": "assistant", "content": generate_welcome_message()}]
# Sidebar: List all conversations
with st.sidebar:
st.title("Conversations")
conversations = fetch_all_conversations()
# Display existing conversations in the sidebar
if conversations:
for convo in conversations:
convo_time = datetime.datetime.fromtimestamp(convo["start_time"]).strftime('%Y-%m-%d %H:%M:%S')
# Use unique keys to differentiate buttons
if st.button(f"Conversation started at {convo_time}", key=f"select_{convo['id']}"):
# Switch to the selected conversation
st.session_state.conversation_id = convo["id"]
st.session_state.messages = fetch_conversation_history(convo["id"])
else:
st.write("No previous conversations.")
# Button to start a new conversation
if st.button("Start a New Conversation", key="new_conversation"):
new_conversation_id = start_new_conversation()
if new_conversation_id:
# Set up a new conversation with a fresh ID
st.session_state.conversation_id = new_conversation_id
st.session_state.messages = [{"role": "assistant", "content": generate_welcome_message()}]
# Main UI
st.title("mAICookbook")
# Display chat history
for message in st.session_state.messages:
if message["role"] == "user":
st.chat_message("user").write(message["content"])
else:
st.chat_message("assistant").write(message["content"])
# Input box for user message
if st.session_state.conversation_id:
if user_input := st.chat_input("Ask a question or say something:"):
# Add user message to chat history
st.session_state.messages.append({"role": "user", "content": user_input})
st.chat_message("user").write(user_input)
# Query the FastAPI server
assistant_response = query_fastapi(user_input, st.session_state.conversation_id)
# Add assistant message to chat history
st.session_state.messages.append({"role": "assistant", "content": assistant_response})
st.chat_message("assistant").write(assistant_response)
else:
st.write("Initializing a new conversation...")