Spaces:
Running
Running
| 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(""" | |
| <style> | |
| .main { background: linear-gradient(to right, #1a1a1a, #2d2d2d); color: #fff; } | |
| .stMarkdown { font-family: 'Helvetica Neue', sans-serif; } | |
| .stButton>button { | |
| margin-right: 0.5rem; | |
| } | |
| </style> | |
| """, 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'<a href="data:file/txt;base64,{b64}" download="{os.path.basename(file)}">📂 Download {os.path.basename(file)}</a>' | |
| def speech_synthesis_html(result): | |
| html_code = f""" | |
| <html><body> | |
| <script> | |
| var msg = new SpeechSynthesisUtterance("{result.replace('"', '')}"); | |
| window.speechSynthesis.speak(msg); | |
| </script> | |
| </body></html> | |
| """ | |
| 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'<video width="{w}" controls autoplay muted loop><source src="data:video/mp4;base64,{d}" type="video/mp4"></video>' | |
| else: | |
| return f'<audio controls style="width:{w};"><source src="data:audio/mpeg;base64,{d}" type="audio/mpeg"></audio>' | |
| 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() | |