blind_llm_helper / src /streamlit_app.py
HugoFTorres's picture
Update src/streamlit_app.py
45398cd verified
Raw
History Blame Contribute Delete
6.96 kB
import os
from pathlib import Path
from openai import OpenAI
from github import Github
import streamlit as st
def upload_to_github(file_path: str, repo_name: str, github_token: str, branch: str = "main") -> str:
#Uploads the image inputs to github. If the file exists, updates it, otherwise, creates the file
github_token = github_token.strip()
g = Github(github_token)
user = g.get_user()
repo = user.get_repo(repo_name)
with open(file_path, "rb") as f:
content = f.read()
file_name = os.path.basename(file_path)
try:
existing_file = repo.get_contents(file_name, ref=branch)
repo.update_file(
existing_file.path,
f"Update {file_name}",
content,
existing_file.sha,
branch=branch
)
print(f"Updated {file_name} in repo {repo_name}")
except Exception:
repo.create_file(
file_name,
f"Add {file_name}",
content,
branch=branch
)
return f"https://raw.githubusercontent.com/{repo.owner.login}/{repo_name}/{branch}/{file_name}"
def text_to_speech(text: str, file_path: str):
#Converts text to speech
client_tts = OpenAI(api_key=os.getenv('OPEN_AI_API_KEY'))
with client_tts.audio.speech.with_streaming_response.create(
model="gpt-4o-mini-tts",
voice="alloy",
input= text
) as response:
response.stream_to_file(file_path)
def speech_to_text(file_path: str):
#Converts speech to text
client_stt = OpenAI(api_key=os.getenv('OPEN_AI_API_KEY'))
with open(file_path, "rb") as audio_file:
response = client_stt.audio.transcriptions.create(
model="whisper-1",
file=audio_file,
response_format="text"
)
return response
system_prompt = """"You are a virtual assistant for blind shoppers. Your role is to guide them safely and clearly through a supermarket to find specific items.
Core Principles: Empathy, be kind and supportive; Clarity, Give precise instructions; Safety, Always avoid potential hazards. Rules:
Never assume the user can see. Do not refer to visual cues (e.g., “look for,” “you should see,” etc.); Only describe what is visible in the picture provided.
Shopping List Handling: Greet the user and ask for their shopping list; Once given, treat the list as final; If multiple items are listed, start with the first one (or whichever appears nearest) without asking the user to choose; After locating one item, move directly to the next.
Guidance Workflow
Location Request: Ask the user to take a picture of their current location.
If the picture is unclear, request another from a different angle.
Image Analysis: Describe what’s in front of the user—aisles, shelves, products, and any text.
Directional Guidance: Give step-by-step directions using only steps and turns.
Example: “Take 15 steps forward, then turn left.”
Never tell the user to turn around or walk backward.
Exclusive Support
You are the user’s only source of assistance.
Do not suggest asking store staff for help.
If you find an item, provide information on the items in display. Any additional information on how to pick an item is not necessary.
Safety Checks
Warn the user if any obstacle or hazard is visible in the path."""
class LLMChat:
def __init__(self,system_prompt=system_prompt,model="gpt-5"):
self.model = model
self.system_prompt = system_prompt
self.history = [{"role":"system","content":system_prompt}]
def llm_call(self, user_prompt, temperature=1):
client = OpenAI(
api_key=os.getenv('OPEN_AI_API_KEY'),
)
self.history.append({"role": "user", "content": user_prompt})
response = client.chat.completions.create(
messages=self.history,
model=self.model,
temperature=temperature,
)
llm_response = str(response.choices[0].message.content).replace("```","").replace("```json","")
self.history.append({"role": "assistant", "content": llm_response})
return llm_response
if 'chat' not in st.session_state:
chat = LLMChat()
st.session_state.chat = chat
else:
chat = st.session_state.chat
speech_file_path = "audio_support_file.mp3"
st.set_page_config(page_title='Shopping Helper', page_icon='🛒')
st.title('Shopping Helper')
#Ensure intro message only autoplays the first time
if 'intro_autoplay' not in st.session_state:
text_to_speech('Hello! What is on your shopping list today?', speech_file_path)
audio_bytes = open(speech_file_path, 'rb').read()
st.audio(audio_bytes, format='audio/mp3', autoplay=True)
st.session_state.intro_autoplay = True
else:
text_to_speech('Hello! What is on your shopping list today?', speech_file_path)
audio_bytes = open(speech_file_path, 'rb').read()
st.audio(audio_bytes, format='audio/mp3', autoplay=False)
audio_shopping_list = st.audio_input(label='Shopping List')
if audio_shopping_list is not None: #Receive shopping list input
audio_bytes = audio_shopping_list.read()
with open(speech_file_path, "wb") as f:
f.write(audio_bytes)
shopping_list = speech_to_text(speech_file_path)
if audio_shopping_list is not None and shopping_list != '':
if 'shopping_list_response_audio' not in st.session_state:
content = [{"type": "text","text": f"Hello! I need help finding {shopping_list}."}]
response = chat.llm_call(content)
text_to_speech(response, speech_file_path)
st.markdown(response)
shopping_list_response = response
audio_bytes = open(speech_file_path, 'rb').read()
st.audio(audio_bytes, format='audio/mp3', autoplay=True)
st.session_state.shopping_list_response = shopping_list_response
st.session_state.shopping_list_response_audio = audio_bytes
else:
st.markdown(st.session_state.shopping_list_response)
st.audio(st.session_state.shopping_list_response_audio, format='audio/mp3', autoplay=False)
image_file = st.file_uploader("Upload a picture of your current position", type=['png','jpg','jpeg'])
if image_file is not None:
st.image(image_file)
b = image_file.getvalue()
with open(image_file.name, "wb") as f:
f.write(b)
image_data_url = upload_to_github(image_file.name, "data-analysis", os.getenv('GITHUB_ACCESS_TOKEN'), branch = 'llm-helper-pictures')
content = [{"type": "text","text": f"Here is my current position."},{"type": "image_url","image_url": {"url": f"{image_data_url}"}}]
response = chat.llm_call(content)
text_to_speech(response, speech_file_path)
st.markdown(response)
audio_bytes = open(speech_file_path, 'rb').read()
st.audio(audio_bytes, format='audio/mp3', autoplay=True)