-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathstreamlit.py
162 lines (131 loc) Β· 5.55 KB
/
streamlit.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
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
import streamlit as st
from dataclasses import dataclass
from typing import List, Optional
import asyncio
from pathlib import Path
import tempfile
import requests
import yaml
from voice_assistant import VoiceAssistant
from components import StatementDisplay
RAG_URL = "http://127.0.0.1:5000/post-data"
@dataclass
class Message:
content: str
role: str
@dataclass
class Statement:
text: str
votes: int = 0
id: str = ""
class RAGService:
"""RAG implementation"""
def get_similar_statements(self, query: str) -> List[str]:
response = requests.post(RAG_URL, params={"text": query}, timeout=10)
return response.json()
class SessionState:
def __init__(self):
if "messages" not in st.session_state:
st.session_state.messages = []
if "current_suggestions" not in st.session_state:
st.session_state.current_suggestions = []
if "recording_active" not in st.session_state:
st.session_state.recording_active = False
if "awaiting_confirmation" not in st.session_state:
st.session_state.awaiting_confirmation = False
class MultimodalPolisApp:
def __init__(self):
self.state = SessionState()
self.voice_assistant = VoiceAssistant(
api_key=st.secrets.get("openrouter_api_key"),
tts_model_path="/home/besher/Documents/tools/piper/en_US-ljspeech-high.onnx",
tts_config_path="/home/besher/Documents/tools/piper/en_en_US_ljspeech_high_en_US-ljspeech-high.onnx.json",
)
self.rag_service = RAGService()
self.statement_display = StatementDisplay()
async def process_text_input(self, text_input: str) -> None:
"""Process text input and generate response"""
messages = [
Message(role="system", content="You are a helpful assistant."),
Message(role="user", content=text_input),
]
response = await self.voice_assistant.llm_client.get_response(messages)
if response:
st.session_state.messages.append({"role": "assistant", "content": response})
similar_statements = self.rag_service.get_similar_statements(text_input)
st.session_state.current_suggestions = [
Statement(text=stmt) for stmt in similar_statements
]
async def handle_voice_interaction(self) -> None:
progress_container = st.empty()
with progress_container:
final_statement, state = await self.voice_assistant.process_voice_input()
# Visual feedback based on state
if state == AssistantState.LISTENING:
st.info("π€ " + state.value, icon="π€")
elif state == AssistantState.PROCESSING:
st.warning("βοΈ " + state.value, icon="βοΈ")
elif state == AssistantState.VERIFYING:
st.info("β " + state.value, icon="β")
elif state == AssistantState.COMPLETED:
st.success("β
" + state.value, icon="β
")
# Process final statement
if final_statement:
similar_statements = self.rag_service.get_similar_statements(
final_statement
)
st.session_state.current_suggestions = [
Statement(text=stmt["text"], id=stmt["hit_id"])
for stmt in similar_statements
]
st.session_state.messages.append(
{"role": "user", "content": final_statement}
)
else:
st.error("β " + state.value, icon="β")
def render_suggestions(self):
"""Render suggestion cards with voting options"""
if st.session_state.current_suggestions:
st.markdown("### Similar Statements")
st.markdown(
"If any of these statements match your intent, you can upvote them:"
)
for idx, suggestion in enumerate(st.session_state.current_suggestions):
col1, col2 = st.columns([4, 1])
with col1:
st.write(suggestion.text)
with col2:
if st.button("β¬οΈ Upvote", key=f"upvote_{idx}"):
suggestion.votes += 1
st.success("Vote recorded!")
st.session_state.current_suggestions = []
st.session_state.recording_active = False
def render_chat_history(self):
"""Render chat history"""
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.write(message["content"])
def render_main_interface(self):
"""Render main application interface"""
st.title("Multimodal Polis π")
# Input section
col1, col2 = st.columns([4, 1])
with col1:
text_input = st.text_input("Type your statement:", key="text_input")
with col2:
voice_button = st.button("π€ Voice Input")
# Process inputs
if text_input:
asyncio.run(self.process_text_input(text_input))
if voice_button:
st.session_state.recording_active = True
asyncio.run(self.handle_voice_interaction())
# Display suggestions and chat history
self.render_suggestions()
self.render_chat_history()
def run(self):
"""Main application entry point"""
self.render_main_interface()
if __name__ == "__main__":
app = MultimodalPolisApp()
app.run()