import streamlit as st
import anthropic, openai, base64, cv2, glob, json, math, os, pytz, random, re, requests, textract, time, zipfile
import plotly.graph_objects as go
import streamlit.components.v1 as components
from datetime import datetime
from audio_recorder_streamlit import audio_recorder
from bs4 import BeautifulSoup
from collections import defaultdict, deque
from dotenv import load_dotenv
from gradio_client import Client
from huggingface_hub import InferenceClient
from io import BytesIO
from PIL import Image
from PyPDF2 import PdfReader
from urllib.parse import quote
from xml.etree import ElementTree as ET
from openai import OpenAI
import extra_streamlit_components as stx
from streamlit.runtime.scriptrunner import get_script_run_ctx
import asyncio
import edge_tts
# 🔧 Config & Setup
st.set_page_config(
page_title="🚲BikeAI🏆 Claude/GPT Research",
page_icon="🚲🏆",
layout="wide",
initial_sidebar_state="auto",
menu_items={
'Get Help': 'https://huggingface.co/awacke1',
'Report a bug': 'https://huggingface.co/spaces/awacke1',
'About': "🚲BikeAI🏆 Claude/GPT Research AI"
}
)
load_dotenv()
openai.api_key = os.getenv('OPENAI_API_KEY') or st.secrets['OPENAI_API_KEY']
anthropic_key = os.getenv("ANTHROPIC_API_KEY_3") or st.secrets["ANTHROPIC_API_KEY"]
claude_client = anthropic.Anthropic(api_key=anthropic_key)
openai_client = OpenAI(api_key=openai.api_key, organization=os.getenv('OPENAI_ORG_ID'))
HF_KEY = os.getenv('HF_KEY')
API_URL = os.getenv('API_URL')
# Session states
st.session_state.setdefault('transcript_history', [])
st.session_state.setdefault('chat_history', [])
st.session_state.setdefault('openai_model', "gpt-4o-2024-05-13")
st.session_state.setdefault('messages', [])
st.session_state.setdefault('last_voice_input', "")
st.session_state.setdefault('editing_file', None)
st.session_state.setdefault('edit_new_name', "")
st.session_state.setdefault('edit_new_content', "")
st.session_state.setdefault('should_rerun', False) # Flag to indicate we need to rerun after operations
# 🎨 Minimal Custom CSS
st.markdown("""
""", unsafe_allow_html=True)
FILE_EMOJIS = {
"md": "📝",
"mp3": "🎵",
}
def generate_filename(prompt, file_type="md"):
ctz = pytz.timezone('US/Central')
date_str = datetime.now(ctz).strftime("%m%d_%H%M")
safe = re.sub(r'[<>:"/\\\\|?*\n]', ' ', prompt)
safe = re.sub(r'\s+', ' ', safe).strip()[:90]
return f"{date_str}_{safe}.{file_type}"
def create_file(filename, prompt, response):
with open(filename, 'w', encoding='utf-8') as f:
f.write(prompt + "\n\n" + response)
st.session_state.should_rerun = True
def get_download_link(file):
with open(file, "rb") as f:
b64 = base64.b64encode(f.read()).decode()
return f'📂 Download {os.path.basename(file)}'
@st.cache_resource
def speech_synthesis_html(result):
html_code = f"""
"""
components.html(html_code, height=0)
async def edge_tts_generate_audio(text, voice="en-US-AriaNeural", rate=0, pitch=0):
if not text.strip():
return None
rate_str = f"{rate:+d}%"
pitch_str = f"{pitch:+d}Hz"
communicate = edge_tts.Communicate(text, voice, rate=rate_str, pitch=pitch_str)
out_fn = generate_filename(text,"mp3")
await communicate.save(out_fn)
st.session_state.should_rerun = True
return out_fn
def speak_with_edge_tts(text, voice="en-US-AriaNeural", rate=0, pitch=0):
return asyncio.run(edge_tts_generate_audio(text, voice, rate, pitch))
def play_and_download_audio(file_path):
if file_path and os.path.exists(file_path):
st.audio(file_path)
st.markdown(get_download_link(file_path), unsafe_allow_html=True)
def process_image(image_path, user_prompt):
with open(image_path, "rb") as imgf:
image_data = imgf.read()
b64img = base64.b64encode(image_data).decode("utf-8")
resp = openai_client.chat.completions.create(
model=st.session_state["openai_model"],
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": [
{"type": "text", "text": user_prompt},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64img}"}}
]}
],
temperature=0.0,
)
return resp.choices[0].message.content
def process_audio(audio_path):
with open(audio_path, "rb") as f:
transcription = openai_client.audio.transcriptions.create(model="whisper-1", file=f)
st.session_state.messages.append({"role": "user", "content": transcription.text})
st.session_state.should_rerun = True
return transcription.text
def process_video(video_path, seconds_per_frame=1):
vid = cv2.VideoCapture(video_path)
total = int(vid.get(cv2.CAP_PROP_FRAME_COUNT))
fps = vid.get(cv2.CAP_PROP_FPS)
skip = int(fps*seconds_per_frame)
frames_b64 = []
for i in range(0, total, skip):
vid.set(cv2.CAP_PROP_POS_FRAMES, i)
ret, frame = vid.read()
if not ret: break
_, buf = cv2.imencode(".jpg", frame)
frames_b64.append(base64.b64encode(buf).decode("utf-8"))
vid.release()
return frames_b64
def process_video_with_gpt(video_path, prompt):
frames = process_video(video_path)
resp = openai_client.chat.completions.create(
model=st.session_state["openai_model"],
messages=[
{"role":"system","content":"Analyze video frames."},
{"role":"user","content":[
{"type":"text","text":prompt},
*[{"type":"image_url","image_url":{"url":f"data:image/jpeg;base64,{fr}"}} for fr in frames]
]}
]
)
return resp.choices[0].message.content
def search_arxiv(query):
st.write("🔍 Searching ArXiv...")
client = Client("awacke1/Arxiv-Paper-Search-And-QA-RAG-Pattern")
r1 = client.predict(prompt=query, llm_model_picked="mistralai/Mixtral-8x7B-Instruct-v0.1", stream_outputs=True, api_name="/ask_llm")
st.markdown("### Mistral-8x7B-Instruct-v0.1 Result")
st.markdown(r1)
r2 = client.predict(prompt=query, llm_model_picked="mistralai/Mistral-7B-Instruct-v0.2", stream_outputs=True, api_name="/ask_llm")
st.markdown("### Mistral-7B-Instruct-v0.2 Result")
st.markdown(r2)
return f"{r1}\n\n{r2}"
def perform_ai_lookup(q, vocal_summary=True, extended_refs=False, titles_summary=True):
start = time.time()
client = Client("awacke1/Arxiv-Paper-Search-And-QA-RAG-Pattern")
r = client.predict(q,20,"Semantic Search","mistralai/Mixtral-8x7B-Instruct-v0.1",api_name="/update_with_rag_md")
refs = r[0]
r2 = client.predict(q,"mistralai/Mixtral-8x7B-Instruct-v0.1",True,api_name="/ask_llm")
result = f"### 🔎 {q}\n\n{r2}\n\n{refs}"
st.markdown(result)
if vocal_summary:
audio_file_main = speak_with_edge_tts(r2, voice="en-US-AriaNeural", rate=0, pitch=0)
st.write("### 🎙️ Vocal Summary (Short Answer)")
play_and_download_audio(audio_file_main)
if extended_refs:
summaries_text = "Here are the summaries from the references: " + refs.replace('"','')
audio_file_refs = speak_with_edge_tts(summaries_text, voice="en-US-AriaNeural", rate=0, pitch=0)
st.write("### 📜 Extended References & Summaries")
play_and_download_audio(audio_file_refs)
if titles_summary:
titles = []
for line in refs.split('\n'):
m = re.search(r"\[([^\]]+)\]", line)
if m:
titles.append(m.group(1))
if titles:
titles_text = "Here are the titles of the papers: " + ", ".join(titles)
audio_file_titles = speak_with_edge_tts(titles_text, voice="en-US-AriaNeural", rate=0, pitch=0)
st.write("### 🔖 Paper Titles")
play_and_download_audio(audio_file_titles)
elapsed = time.time()-start
st.write(f"**Total Elapsed:** {elapsed:.2f} s")
fn = generate_filename(q,"md")
create_file(fn,q,result)
return result
def process_with_gpt(text):
if not text: return
st.session_state.messages.append({"role":"user","content":text})
with st.chat_message("user"):
st.markdown(text)
with st.chat_message("assistant"):
c = openai_client.chat.completions.create(
model=st.session_state["openai_model"],
messages=st.session_state.messages,
stream=False
)
ans = c.choices[0].message.content
st.write("GPT-4o: " + ans)
create_file(generate_filename(text,"md"),text,ans)
st.session_state.messages.append({"role":"assistant","content":ans})
return ans
def process_with_claude(text):
if not text: return
with st.chat_message("user"):
st.markdown(text)
with st.chat_message("assistant"):
r = claude_client.messages.create(
model="claude-3-sonnet-20240229",
max_tokens=1000,
messages=[{"role":"user","content":text}]
)
ans = r.content[0].text
st.write("Claude: " + ans)
create_file(generate_filename(text,"md"),text,ans)
st.session_state.chat_history.append({"user":text,"claude":ans})
return ans
def create_zip_of_files():
md_files = glob.glob("*.md")
mp3_files = glob.glob("*.mp3")
all_files = md_files + mp3_files
zip_name = "all_files.zip"
with zipfile.ZipFile(zip_name,'w') as z:
for f in all_files:
z.write(f)
st.session_state.should_rerun = True
return zip_name
def get_media_html(p,typ="video",w="100%"):
d = base64.b64encode(open(p,'rb').read()).decode()
if typ=="video":
return f''
else:
return f''
def load_files_for_sidebar():
# Gather all md and mp3 files
md_files = glob.glob("*.md")
mp3_files = glob.glob("*.mp3")
files_by_ext = defaultdict(list)
for f in md_files:
files_by_ext['md'].append(f)
for f in mp3_files:
files_by_ext['mp3'].append(f)
# Sort each extension group by modification time descending
for ext in files_by_ext:
files_by_ext[ext].sort(key=lambda x: os.path.getmtime(x), reverse=True)
return files_by_ext
def display_file_manager_sidebar(files_by_ext):
st.sidebar.title("🎵 Audio & Document Manager")
md_files = files_by_ext.get('md', [])
mp3_files = files_by_ext.get('mp3', [])
# Delete all buttons
col_del = st.sidebar.columns(2)
with col_del[0]:
if st.button("🗑 Delete All MD"):
for f in md_files:
os.remove(f)
st.session_state.should_rerun = True
with col_del[1]:
if st.button("🗑 Delete All MP3"):
for f in mp3_files:
os.remove(f)
st.session_state.should_rerun = True
# Sort extensions by number of files descending
ext_counts = {ext: len(files) for ext, files in files_by_ext.items()}
sorted_ext = sorted(files_by_ext.keys(), key=lambda x: ext_counts[x], reverse=True)
# Display groups
for ext in sorted_ext:
emoji = FILE_EMOJIS.get(ext, "📦")
count = len(files_by_ext[ext])
with st.sidebar.expander(f"{emoji} {ext.upper()} Files ({count})"):
for f in files_by_ext[ext]:
fname = os.path.basename(f)
ctime = datetime.fromtimestamp(os.path.getmtime(f)).strftime("%Y-%m-%d %H:%M:%S")
col1, col2, col3, col4 = st.columns([2,1,1,1])
with col1:
st.write(f"**{fname}** - {ctime}")
with col2:
# View button
if ext == "md":
if st.button("👀", key="view_"+f):
content = open(f,'r',encoding='utf-8').read()
st.write("**Viewing file content:**")
st.markdown(content)
else: # mp3
if st.button("👀", key="view_"+f):
st.write(f"Playing: {fname}")
st.audio(f)
with col3:
# Edit button for MD
if ext == "md":
if st.button("✏️", key="edit_"+f):
st.session_state.editing_file = f
st.session_state.edit_new_name = fname.replace(".md","")
st.session_state.edit_new_content = open(f,'r',encoding='utf-8').read()
st.session_state.should_rerun = True
else:
pass
with col4:
# Delete button
if st.button("🗑", key="del_"+f):
os.remove(f)
st.session_state.should_rerun = True
# Download all as zip
if (len(md_files) > 0 or len(mp3_files) > 0) and st.sidebar.button("⬇️ Download All (.md and .mp3)"):
z = create_zip_of_files()
st.sidebar.markdown(get_download_link(z),unsafe_allow_html=True)
# If editing an md file
if st.session_state.editing_file and os.path.exists(st.session_state.editing_file):
st.sidebar.subheader(f"Editing: {os.path.basename(st.session_state.editing_file)}")
st.session_state.edit_new_name = st.sidebar.text_input("New name (without extension):", value=st.session_state.edit_new_name)
st.session_state.edit_new_content = st.sidebar.text_area("Content:", st.session_state.edit_new_content, height=200)
c1,c2 = st.sidebar.columns(2)
with c1:
if st.button("Save Changes"):
old_path = st.session_state.editing_file
new_path = st.session_state.edit_new_name + ".md"
if new_path != os.path.basename(old_path):
os.rename(old_path, new_path)
with open(new_path,'w',encoding='utf-8') as f:
f.write(st.session_state.edit_new_content)
st.session_state.editing_file = None
st.session_state.should_rerun = True
with c2:
if st.button("Cancel"):
st.session_state.editing_file = None
st.session_state.should_rerun = True
def main():
st.sidebar.markdown("### 🚲BikeAI🏆 Multi-Agent Research AI")
tab_main = st.radio("Action:",["🎤 Voice Input","📸 Media Gallery","🔍 Search ArXiv","📝 File Editor"],horizontal=True)
model_choice = st.sidebar.radio("AI Model:", ["Arxiv","GPT-4o","Claude-3","GPT+Claude+Arxiv"], index=0)
# Main Input Component
mycomponent = components.declare_component("mycomponent", path="mycomponent")
val = mycomponent(my_input_value="Hello")
if val:
user_input = val.strip()
if user_input:
if model_choice == "GPT-4o":
process_with_gpt(user_input)
elif model_choice == "Claude-3":
process_with_claude(user_input)
elif model_choice == "Arxiv":
st.subheader("Arxiv Only Results:")
perform_ai_lookup(user_input, vocal_summary=True, extended_refs=False, titles_summary=True)
else:
col1,col2,col3=st.columns(3)
with col1:
st.subheader("GPT-4o Omni:")
try: process_with_gpt(user_input)
except: st.write('GPT 4o error')
with col2:
st.subheader("Claude-3 Sonnet:")
try: process_with_claude(user_input)
except: st.write('Claude error')
with col3:
st.subheader("Arxiv + Mistral:")
try:
perform_ai_lookup(user_input, vocal_summary=True, extended_refs=False, titles_summary=True)
except:
st.write("Arxiv error")
if tab_main == "🔍 Search ArXiv":
st.subheader("🔍 Search ArXiv")
q=st.text_input("Research query:")
# 🎛️ Audio Generation Options
st.markdown("### 🎛️ Audio Generation Options")
vocal_summary = st.checkbox("🎙️ Vocal Summary (Short Answer)", value=True)
extended_refs = st.checkbox("📜 Extended References & Summaries (Long)", value=False)
titles_summary = st.checkbox("🔖 Paper Titles Only", value=True)
if q:
q = q.strip()
if q and st.button("Run ArXiv Query"):
r = perform_ai_lookup(q, vocal_summary=vocal_summary, extended_refs=extended_refs, titles_summary=titles_summary)
st.markdown(r)
elif tab_main == "🎤 Voice Input":
st.subheader("🎤 Voice Recognition")
user_text = st.text_area("Message:", height=100)
user_text = user_text.strip()
if st.button("Send 📨"):
if user_text:
if model_choice == "GPT-4o":
process_with_gpt(user_text)
elif model_choice == "Claude-3":
process_with_claude(user_text)
elif model_choice == "Arxiv":
st.subheader("Arxiv Only Results:")
perform_ai_lookup(user_text, vocal_summary=True, extended_refs=False, titles_summary=True)
else:
col1,col2,col3=st.columns(3)
with col1:
st.subheader("GPT-4o Omni:")
process_with_gpt(user_text)
with col2:
st.subheader("Claude-3 Sonnet:")
process_with_claude(user_text)
with col3:
st.subheader("Arxiv & Mistral:")
res = perform_ai_lookup(user_text, vocal_summary=True, extended_refs=False, titles_summary=True)
st.markdown(res)
st.subheader("📜 Chat History")
t1,t2=st.tabs(["Claude History","GPT-4o History"])
with t1:
for c in st.session_state.chat_history:
st.write("**You:**", c["user"])
st.write("**Claude:**", c["claude"])
with t2:
for m in st.session_state.messages:
with st.chat_message(m["role"]):
st.markdown(m["content"])
elif tab_main == "📸 Media Gallery":
st.header("🎬 Media Gallery - Images and Videos")
tabs = st.tabs(["🖼️ Images", "🎥 Video"])
with tabs[0]:
imgs = glob.glob("*.png")+glob.glob("*.jpg")
if imgs:
c = st.slider("Cols",1,5,3)
cols = st.columns(c)
for i,f in enumerate(imgs):
with cols[i%c]:
st.image(Image.open(f),use_container_width=True)
if st.button(f"👀 Analyze {os.path.basename(f)}", key=f"analyze_{f}"):
a = process_image(f,"Describe this image.")
st.markdown(a)
else:
st.write("No images found.")
with tabs[1]:
vids = glob.glob("*.mp4")
if vids:
for v in vids:
with st.expander(f"🎥 {os.path.basename(v)}"):
st.markdown(get_media_html(v,"video"),unsafe_allow_html=True)
if st.button(f"Analyze {os.path.basename(v)}", key=f"analyze_{v}"):
a = process_video_with_gpt(v,"Describe video.")
st.markdown(a)
else:
st.write("No videos found.")
elif tab_main == "📝 File Editor":
# Existing code for inline editing if needed
if getattr(st.session_state,'current_file',None):
st.subheader(f"Editing: {st.session_state.current_file}")
new_text = st.text_area("Content:", st.session_state.file_content, height=300)
if st.button("Save"):
with open(st.session_state.current_file,'w',encoding='utf-8') as f:
f.write(new_text)
st.success("Updated!")
st.session_state.should_rerun = True
else:
st.write("Select a file from the sidebar to edit.")
# After all main content is processed, load files and display in sidebar
files_by_ext = load_files_for_sidebar()
display_file_manager_sidebar(files_by_ext)
# If we performed an operation, rerun now at the end
if st.session_state.should_rerun:
st.session_state.should_rerun = False
st.rerun()
if __name__=="__main__":
main()