import streamlit as st
import anthropic, openai, base64, cv2, glob, json, math, os, pytz, random, re, requests, 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
# π― 1. Core Configuration & 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()
# π 2. API Setup & Clients
openai_api_key = os.getenv('OPENAI_API_KEY', "")
anthropic_key = os.getenv('ANTHROPIC_API_KEY_3', "")
xai_key = os.getenv('xai',"")
if 'OPENAI_API_KEY' in st.secrets:
openai_api_key = st.secrets['OPENAI_API_KEY']
if 'ANTHROPIC_API_KEY' in st.secrets:
anthropic_key = st.secrets["ANTHROPIC_API_KEY"]
openai.api_key = openai_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')
# π 3. Session State Management
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_prefix' not in st.session_state:
st.session_state['viewing_prefix'] = None
if 'should_rerun' not in st.session_state:
st.session_state['should_rerun'] = False
if 'old_val' not in st.session_state:
st.session_state['old_val'] = None
# π¨ 4. Custom CSS
st.markdown("""
""", unsafe_allow_html=True)
FILE_EMOJIS = {
"md": "π",
"mp3": "π΅",
}
def clean_for_speech(text: str) -> str:
"""Clean text for speech synthesis"""
text = text.replace("\n", " ")
text = text.replace("", " ")
text = text.replace("#", "")
text = re.sub(r"\(https?:\/\/[^\)]+\)", "", text)
text = re.sub(r"\s+", " ", text).strip()
return text
@st.cache_resource
def speech_synthesis_html(result):
"""Create HTML for speech synthesis"""
html_code = f"""
"""
components.html(html_code, height=0)
async def edge_tts_generate_audio(text, voice="en-US-AriaNeural", rate=0, pitch=0):
"""Generate audio using Edge TTS"""
text = clean_for_speech(text)
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, text, "mp3")
await communicate.save(out_fn)
return out_fn
def speak_with_edge_tts(text, voice="en-US-AriaNeural", rate=0, pitch=0):
"""Wrapper for edge TTS generation"""
return asyncio.run(edge_tts_generate_audio(text, voice, rate, pitch))
def play_and_download_audio(file_path):
"""Play and provide download link for audio"""
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 save_full_transcript(query, text):
"""Save full transcript of Arxiv results as a file."""
create_file(query, text, "md")
def perform_ai_lookup(q, vocal_summary=True, extended_refs=False, titles_summary=True, full_audio=False):
"""Perform Arxiv search and generate audio summaries"""
start = time.time()
client = Client("awacke1/Arxiv-Paper-Search-And-QA-RAG-Pattern")
refs = client.predict(q,20,"Semantic Search","mistralai/Mixtral-8x7B-Instruct-v0.1",api_name="/update_with_rag_md")[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)
# Generate full audio version if requested
if full_audio:
complete_text = f"Complete response for query: {q}. {clean_for_speech(r2)} {clean_for_speech(refs)}"
audio_file_full = speak_with_edge_tts(complete_text)
st.write("### π Full Audio")
play_and_download_audio(audio_file_full)
if vocal_summary:
main_text = clean_for_speech(r2)
audio_file_main = speak_with_edge_tts(main_text)
st.write("### π Short Audio")
play_and_download_audio(audio_file_main)
if extended_refs:
summaries_text = "Extended references: " + refs.replace('"','')
summaries_text = clean_for_speech(summaries_text)
audio_file_refs = speak_with_edge_tts(summaries_text)
st.write("### π Long Refs")
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 = "Titles: " + ", ".join(titles)
titles_text = clean_for_speech(titles_text)
audio_file_titles = speak_with_edge_tts(titles_text)
st.write("### π Titles")
play_and_download_audio(audio_file_titles)
elapsed = time.time()-start
st.write(f"**Total Elapsed:** {elapsed:.2f} s")
# Create file with result
create_file(q, result, "md")
return result
def process_with_gpt(text):
"""Process text with GPT-4"""
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(text, ans, "md")
st.session_state.messages.append({"role":"assistant","content":ans})
return ans
def process_with_claude(text):
"""Process text with Claude"""
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-3.5: " + ans)
create_file(text, ans, "md")
st.session_state.chat_history.append({"user":text,"claude":ans})
return ans
def generate_filename(prompt, response, file_type="md"):
"""Generate filename with timestamp and cleaned text."""
timestamp = datetime.now().strftime("%y%m_%H%M")
safe_text = re.sub(r'[^\w\s-]', '', prompt[:50])
return f"{timestamp}_{safe_text}.{file_type}"
def create_file(prompt, response, file_type="md"):
"""Create file with content."""
filename = generate_filename(prompt.strip(), response.strip(), file_type)
with open(filename, 'w', encoding='utf-8') as f:
f.write(prompt + "\n\n" + response)
return filename
def get_download_link(file):
"""Generate download link for file"""
with open(file, "rb") as f:
b64 = base64.b64encode(f.read()).decode()
return f'π Download {os.path.basename(file)}'
import streamlit as st
import anthropic, openai, base64, cv2, glob, json, math, os, pytz, random, re, requests, 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
# π― 1. Core Configuration & 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()
# π 2. API Setup & Clients
openai_api_key = os.getenv('OPENAI_API_KEY', "")
anthropic_key = os.getenv('ANTHROPIC_API_KEY_3', "")
xai_key = os.getenv('xai',"")
if 'OPENAI_API_KEY' in st.secrets:
openai_api_key = st.secrets['OPENAI_API_KEY']
if 'ANTHROPIC_API_KEY' in st.secrets:
anthropic_key = st.secrets["ANTHROPIC_API_KEY"]
openai.api_key = openai_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')
# π 3. Session State Management
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_prefix' not in st.session_state:
st.session_state['viewing_prefix'] = None
if 'should_rerun' not in st.session_state:
st.session_state['should_rerun'] = False
if 'old_val' not in st.session_state:
st.session_state['old_val'] = None
# π¨ 4. Custom CSS
st.markdown("""
""", unsafe_allow_html=True)
FILE_EMOJIS = {
"md": "π",
"mp3": "π΅",
}
def clean_for_speech(text: str) -> str:
"""Clean text for speech synthesis"""
text = text.replace("\n", " ")
text = text.replace("", " ")
text = text.replace("#", "")
text = re.sub(r"\(https?:\/\/[^\)]+\)", "", text)
text = re.sub(r"\s+", " ", text).strip()
return text
@st.cache_resource
def speech_synthesis_html(result):
"""Create HTML for speech synthesis"""
html_code = f"""
"""
components.html(html_code, height=0)
async def edge_tts_generate_audio(text, voice="en-US-AriaNeural", rate=0, pitch=0):
"""Generate audio using Edge TTS"""
text = clean_for_speech(text)
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, text, "mp3")
await communicate.save(out_fn)
return out_fn
def speak_with_edge_tts(text, voice="en-US-AriaNeural", rate=0, pitch=0):
"""Wrapper for edge TTS generation"""
return asyncio.run(edge_tts_generate_audio(text, voice, rate, pitch))
def play_and_download_audio(file_path):
"""Play and provide download link for audio"""
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 save_full_transcript(query, text):
"""Save full transcript of Arxiv results as a file."""
create_file(query, text, "md")
def perform_ai_lookup(q, vocal_summary=True, extended_refs=False, titles_summary=True, full_audio=False):
"""Perform Arxiv search and generate audio summaries"""
start = time.time()
client = Client("awacke1/Arxiv-Paper-Search-And-QA-RAG-Pattern")
refs = client.predict(q,20,"Semantic Search","mistralai/Mixtral-8x7B-Instruct-v0.1",api_name="/update_with_rag_md")[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)
# Generate full audio version if requested
if full_audio:
complete_text = f"Complete response for query: {q}. {clean_for_speech(r2)} {clean_for_speech(refs)}"
audio_file_full = speak_with_edge_tts(complete_text)
st.write("### π Full Audio")
play_and_download_audio(audio_file_full)
if vocal_summary:
main_text = clean_for_speech(r2)
audio_file_main = speak_with_edge_tts(main_text)
st.write("### π Short Audio")
play_and_download_audio(audio_file_main)
if extended_refs:
summaries_text = "Extended references: " + refs.replace('"','')
summaries_text = clean_for_speech(summaries_text)
audio_file_refs = speak_with_edge_tts(summaries_text)
st.write("### π Long Refs")
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 = "Titles: " + ", ".join(titles)
titles_text = clean_for_speech(titles_text)
audio_file_titles = speak_with_edge_tts(titles_text)
st.write("### π Titles")
play_and_download_audio(audio_file_titles)
elapsed = time.time()-start
st.write(f"**Total Elapsed:** {elapsed:.2f} s")
# Create file with result
create_file(q, result, "md")
return result
def process_with_gpt(text):
"""Process text with GPT-4"""
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(text, ans, "md")
st.session_state.messages.append({"role":"assistant","content":ans})
return ans
def process_with_claude(text):
"""Process text with Claude"""
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-3.5: " + ans)
create_file(text, ans, "md")
st.session_state.chat_history.append({"user":text,"claude":ans})
return ans
def generate_filename(prompt, response, file_type="md"):
"""Generate filename with timestamp and cleaned text."""
timestamp = datetime.now().strftime("%y%m_%H%M")
safe_text = re.sub(r'[^\w\s-]', '', prompt[:50])
return f"{timestamp}_{safe_text}.{file_type}"
def create_file(prompt, response, file_type="md"):
"""Create file with content."""
filename = generate_filename(prompt.strip(), response.strip(), file_type)
with open(filename, 'w', encoding='utf-8') as f:
f.write(prompt + "\n\n" + response)
return filename
def get_download_link(file):
"""Generate download link for file"""
with open(file, "rb") as f:
b64 = base64.b64encode(f.read()).decode()
return f'π Download {os.path.basename(file)}'
def main():
st.sidebar.markdown("### π²BikeAIπ Multi-Agent Research")
tab_main = st.radio("Action:",["π€ Voice","πΈ Media","π ArXiv","π Editor"],horizontal=True)
mycomponent = components.declare_component("mycomponent", path="mycomponent")
val = mycomponent(my_input_value="Hello")
# Show input in text box for editing if detected
if val:
val_stripped = val.replace('\n', ' ')
edited_input = st.text_area("βοΈ Edit Input:", value=val_stripped, height=100)
run_option = st.selectbox("Model:", ["Arxiv", "GPT-4o", "Claude-3.5"])
col1, col2 = st.columns(2)
with col1:
autorun = st.checkbox("β AutoRun", value=True)
with col2:
full_audio = st.checkbox("πFullAudio", value=False,
help="Full audio of results")
full_transcript = st.checkbox("π§ΎFullTranscript", value=False,
help="Generate a full transcript file")
if q and st.button("πRun"):
result = perform_ai_lookup(q, vocal_summary=vocal_summary, extended_refs=extended_refs,
titles_summary=titles_summary, full_audio=full_audio)
if full_transcript:
save_full_transcript(q, result)
st.markdown("### Change Prompt & Re-Run")
q_new = st.text_input("π Modify Query:")
if q_new and st.button("π Re-Run with Modified Query"):
result = perform_ai_lookup(q_new, vocal_summary=vocal_summary, extended_refs=extended_refs,
titles_summary=titles_summary, full_audio=full_audio)
if full_transcript:
save_full_transcript(q_new, result)
elif tab_main == "π€ Voice":
st.subheader("π€ Voice Input")
user_text = st.text_area("π¬ Message:", height=100)
user_text = user_text.strip().replace('\n', ' ')
if st.button("π¨ Send"):
process_with_gpt(user_text)
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":
st.header("πΈ Images & π₯ 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.video(v)
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 == "π 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.")
groups, sorted_prefixes = load_files_for_sidebar()
display_file_manager_sidebar(groups, sorted_prefixes)
if st.session_state.viewing_prefix and st.session_state.viewing_prefix in groups:
st.write("---")
st.write(f"**Viewing Group:** {st.session_state.viewing_prefix}")
for f in groups[st.session_state.viewing_prefix]:
fname = os.path.basename(f)
ext = os.path.splitext(fname)[1].lower().strip('.')
st.write(f"### {fname}")
if ext == "md":
content = open(f,'r',encoding='utf-8').read()
st.markdown(content)
elif ext == "mp3":
st.audio(f)
else:
st.markdown(get_download_link(f), unsafe_allow_html=True)
if st.button("β Close"):
st.session_state.viewing_prefix = None
if st.session_state.should_rerun:
st.session_state.should_rerun = False
st.rerun()
if __name__ == "__main__":
main()