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
if 'transcript_history' not in st.session_state:
st.session_state['transcript_history'] = []
if 'chat_history' not in st.session_state:
st.session_state['chat_history'] = []
if 'openai_model' not in st.session_state:
st.session_state['openai_model'] = "gpt-4o-2024-05-13"
if 'messages' not in st.session_state:
st.session_state['messages'] = []
if 'last_voice_input' not in st.session_state:
st.session_state['last_voice_input'] = ""
if 'editing_file' not in st.session_state:
st.session_state['editing_file'] = None
if 'edit_new_name' not in st.session_state:
st.session_state['edit_new_name'] = ""
if 'edit_new_content' not in st.session_state:
st.session_state['edit_new_content'] = ""
if 'viewing_file' not in st.session_state:
st.session_state['viewing_file'] = None
if 'viewing_file_type' not in st.session_state:
st.session_state['viewing_file_type'] = None
if 'should_rerun' not in st.session_state:
st.session_state['should_rerun'] = False
# π¨ 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):
# Creating file does not trigger immediate rerun
with open(filename, 'w', encoding='utf-8') as f:
f.write(prompt + "\n\n" + response)
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):
# Just create mp3 file, no immediate rerun
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)
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)
dl_link = f'Download {os.path.basename(file_path)}'
st.markdown(dl_link, 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})
# No immediate rerun
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, mp3_files):
# Exclude README.md if present
md_files = [f for f in md_files if os.path.basename(f).lower() != 'readme.md']
all_files = md_files + mp3_files
if not all_files:
return None
# Build a descriptive name from file stems
stems = [os.path.splitext(os.path.basename(f))[0] for f in all_files]
# Join them
joined = "_".join(stems)
# Truncate if too long
if len(joined) > 50:
joined = joined[:50] + "_etc"
zip_name = f"{joined}.zip"
with zipfile.ZipFile(zip_name,'w') as z:
for f in all_files:
z.write(f)
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")
# Exclude README.md from listings
md_files = [f for f in md_files if os.path.basename(f).lower() != 'readme.md']
files_by_ext = defaultdict(list)
if md_files: files_by_ext['md'].extend(md_files)
if mp3_files: files_by_ext['mp3'].extend(mp3_files)
# 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', [])
# Buttons to delete all except README.md (already excluded)
col_del = st.sidebar.columns(3)
with col_del[0]:
if st.button("π Del All MD"):
for f in md_files:
os.remove(f)
st.session_state.should_rerun = True
with col_del[1]:
if st.button("π Del All MP3"):
for f in mp3_files:
os.remove(f)
st.session_state.should_rerun = True
with col_del[2]:
if st.button("β¬οΈ Zip All"):
# create a zip of all md and mp3 except README.md
z = create_zip_of_files(md_files, mp3_files)
if z:
st.sidebar.markdown(get_download_link(z),unsafe_allow_html=True)
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 files with actions
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})", expanded=True):
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")
# Show filename and buttons in a row
st.write(f"**{fname}** - {ctime}")
file_buttons_col = st.columns([1,1,1])
with file_buttons_col[0]:
if st.button("πView", key="view_"+f):
st.session_state.viewing_file = f
st.session_state.viewing_file_type = ext
# No rerun needed, just set state
with file_buttons_col[1]:
if ext == "md":
if st.button("βοΈEdit", 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
with file_buttons_col[2]:
if st.button("πDel", key="del_"+f):
os.remove(f)
st.session_state.should_rerun = 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 (no 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"):
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"):
perform_ai_lookup(q, vocal_summary=vocal_summary, extended_refs=extended_refs, titles_summary=titles_summary)
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":
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 main content, load files and display in sidebar
files_by_ext = load_files_for_sidebar()
display_file_manager_sidebar(files_by_ext)
# If viewing a file, show its content below (in the main area)
if st.session_state.viewing_file and os.path.exists(st.session_state.viewing_file):
st.write("---")
st.write(f"**Viewing File:** {os.path.basename(st.session_state.viewing_file)}")
if st.session_state.viewing_file_type == "md":
# show markdown
content = open(st.session_state.viewing_file,'r',encoding='utf-8').read()
st.markdown(content)
elif st.session_state.viewing_file_type == "mp3":
# show audio
st.audio(st.session_state.viewing_file)
# Optionally add a "Close View" button
if st.button("Close View"):
st.session_state.viewing_file = None
st.session_state.viewing_file_type = None
# If user-triggered changes happened, rerun once at the end
if st.session_state.should_rerun:
st.session_state.should_rerun = False
st.rerun()
if __name__=="__main__":
main()