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 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
# 🔧 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')
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', "")
# 🎨 Minimal Custom CSS
st.markdown("""
""", unsafe_allow_html=True)
# 🔑 Common Utilities
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)
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)
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})
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):
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}"
speech_synthesis_html(r2)
st.markdown(result)
elapsed = time.time()-start
st.write(f"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(files):
zip_name = "all_files.zip"
with zipfile.ZipFile(zip_name,'w') as z:
for f in 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 create_media_gallery():
st.header("🎬 Media Gallery")
tabs = st.tabs(["🖼️ Images", "🎵 Audio", "🎥 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)}"):
a = process_image(f,"Describe this image.")
st.markdown(a)
with tabs[1]:
auds = glob.glob("*.mp3")+glob.glob("*.wav")
for a in auds:
with st.expander(f"🎵 {os.path.basename(a)}"):
st.markdown(get_media_html(a,"audio"),unsafe_allow_html=True)
if st.button(f"Transcribe {os.path.basename(a)}"):
t = process_audio(a)
st.write(t)
with tabs[2]:
vids = glob.glob("*.mp4")
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)}"):
a = process_video_with_gpt(v,"Describe video.")
st.markdown(a)
def display_file_manager():
st.sidebar.title("📁 File Management")
files = sorted(glob.glob("*.md"),reverse=True)
if st.sidebar.button("🗑 Delete All"):
for f in files: os.remove(f)
st.experimental_rerun()
if st.sidebar.button("⬇️ Download All"):
z= create_zip_of_files(files)
st.sidebar.markdown(get_download_link(z),unsafe_allow_html=True)
for f in files:
col1,col2,col3,col4 = st.sidebar.columns([1,3,1,1])
with col1:
if st.button("🌐",key="v"+f):
st.session_state.current_file=f
c=open(f,'r',encoding='utf-8').read()
st.write(c)
with col2:
st.markdown(get_download_link(f),unsafe_allow_html=True)
with col3:
if st.button("📂",key="e"+f):
st.session_state.current_file=f
st.session_state.file_content=open(f,'r',encoding='utf-8').read()
with col4:
if st.button("🗑",key="d"+f):
os.remove(f)
st.experimental_rerun()
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:",["GPT+Claude+Arxiv","GPT-4o","Claude-3"])
# Speech-to-Text component placeholder (example)
mycomponent = components.declare_component("mycomponent", path="mycomponent")
val = mycomponent(my_input_value="Hello")
if val:
user_input = val
if model_choice == "GPT-4o":
process_with_gpt(user_input)
elif model_choice == "Claude-3":
process_with_claude(user_input)
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:
r = perform_ai_lookup(user_input)
st.markdown(r)
except:
st.write("Arxiv error")
if tab_main == "🎤 Voice Input":
st.subheader("🎤 Voice Recognition")
user_text = st.text_area("Message:", height=100)
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)
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)
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":
create_media_gallery()
elif tab_main == "🔍 Search ArXiv":
q=st.text_input("Research query:")
if q:
r=search_arxiv(q)
st.markdown(r)
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!")
display_file_manager()
if __name__=="__main__":
main()