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 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 # ensure this is installed (pip install 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') | |
| 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(""" | |
| <style> | |
| .main { background: linear-gradient(to right, #1a1a1a, #2d2d2d); color: #fff; } | |
| .stMarkdown { font-family: 'Helvetica Neue', sans-serif; } | |
| </style> | |
| """, 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'<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) | |
| #------------add EdgeTTS | |
| # --- NEW FUNCTIONS FOR EDGE TTS --- | |
| 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) | |
| 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}) | |
| 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) | |
| # Main Vocal Summary (Short Answer) | |
| if vocal_summary: | |
| start_main_part = time.time() | |
| 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) | |
| st.write(f"**Elapsed (Short Answer):** {time.time() - start_main_part:.2f} s") | |
| # Extended References & Summaries (optional) | |
| if extended_refs: | |
| start_refs_part = time.time() | |
| 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) | |
| st.write(f"**Elapsed (Extended References):** {time.time() - start_refs_part:.2f} s") | |
| # Paper Titles Only (short) | |
| if titles_summary: | |
| start_titles_part = time.time() | |
| 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) | |
| st.write(f"**Elapsed (Titles):** {time.time() - start_titles_part:.2f} s") | |
| 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) | |
| 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>' | |
| # CHANGES START HERE: | |
| # Define file emojis and the directory to scan | |
| FILE_EMOJIS = { | |
| "cards": "🗃️", | |
| "csv": "📊", | |
| "heic": "🖼️", | |
| "ico": "🪧", | |
| "jpeg": "🖼️", | |
| "json": "🔧", | |
| "md": "📝", | |
| "mid": "🎼", | |
| "mov": "🎥", | |
| "mp3": "🎵", | |
| "mp4": "🎞️", | |
| "png": "🖼️", | |
| "svg": "🖌️", | |
| "txt": "📄", | |
| "wav": "🎶", | |
| "webm": "📽️", | |
| "webp": "🏞️", | |
| "zip": "📦", | |
| } | |
| MEDIA_DIR = "Media" | |
| def get_file_ext(filename): | |
| return os.path.splitext(filename)[1].lower().strip('.') | |
| def load_files(): | |
| all_files = [] | |
| for root, dirs, files in os.walk(MEDIA_DIR): | |
| for f in files: | |
| fp = os.path.join(root, f) | |
| if os.path.isfile(fp): | |
| ext = get_file_ext(fp) | |
| mod_time = os.path.getmtime(fp) | |
| all_files.append((fp, ext, mod_time)) | |
| return all_files | |
| def display_files_sidebar(): | |
| st.sidebar.title("📂 Media Files") | |
| all_files = load_files() | |
| from collections import defaultdict | |
| ext_map = defaultdict(list) | |
| for fp, ext, mod_time in all_files: | |
| ext_map[ext].append((fp, mod_time)) | |
| # Sort files in each extension group by modification time descending | |
| for ext in ext_map: | |
| ext_map[ext].sort(key=lambda x: x[1], reverse=True) | |
| # Sort extensions by number of files descending | |
| sorted_ext = sorted(ext_map.keys(), key=lambda x: len(ext_map[x]), reverse=True) | |
| for ext in sorted_ext: | |
| emoji = FILE_EMOJIS.get(ext, "📁") | |
| count = len(ext_map[ext]) | |
| with st.sidebar.expander(f"{emoji} {ext.upper()} ({count})"): | |
| for fp, mod_time in ext_map[ext]: | |
| basename = os.path.basename(fp) | |
| last_mod = datetime.fromtimestamp(mod_time).strftime("%Y-%m-%d %H:%M:%S") | |
| col1, col2 = st.columns([3,1]) | |
| with col1: | |
| st.write(f"**{basename}** - Modified: {last_mod}") | |
| with col2: | |
| if ext == "mp3": | |
| # For MP3, load download link only after user clicks an expander | |
| mp3_exp = st.expander("Load MP3 Download Link") | |
| with mp3_exp: | |
| st.markdown(get_download_link(fp), unsafe_allow_html=True) | |
| # If desired, add an on-demand audio player here. | |
| else: | |
| # Direct download link for other files | |
| st.markdown(get_download_link(fp), unsafe_allow_html=True) | |
| # CHANGES END HERE | |
| 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) | |
| # Declare the 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)}"): | |
| 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)}"): | |
| 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!") | |
| else: | |
| st.write("Select a file from the sidebar to edit.") | |
| # Use the new sidebar display function | |
| display_files_sidebar() | |
| if __name__=="__main__": | |
| main() | |