Spaces:
Runtime error
Runtime error
Download app.py from Ines1994/ENA-Chatbot: direct link, hf CLI and curl.
- Browser
- Download file 16 kB
-
https://huggingface.co/spaces/Ines1994/ENA-Chatbot/resolve/main/app.py
- Command line
-
hf download hf://spaces/Ines1994/ENA-Chatbot/app.py
-
curl -L -o app.py https://huggingface.co/spaces/Ines1994/ENA-Chatbot/resolve/main/app.py
16 kB
| import os | |
| import re | |
| import time | |
| import streamlit as st | |
| from dotenv import load_dotenv | |
| from config import * | |
| from utils import normalize_arabic, detect_lang, format_latency, extract_pdf_text | |
| from rag_engine import ENAEngine | |
| from agentic_rag import AgenticRAG | |
| # โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ | |
| # โ๏ธ INITIALIZATION | |
| # โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ | |
| load_dotenv() | |
| st.set_page_config(page_title="ENA Chatbot PRO", page_icon="๐", layout="wide") | |
| st.markdown(UI_CSS, unsafe_allow_html=True) | |
| def get_engine(): | |
| token = os.getenv("GROQ_TOKEN") or st.secrets.get("GROQ_TOKEN") | |
| return ENAEngine(groq_token=token) | |
| def get_agent(_engine): | |
| return AgenticRAG(_engine) | |
| engine = get_engine() | |
| agent = get_agent(engine) | |
| # โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ | |
| # ๐ ๏ธ UI HELPERS | |
| # โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ | |
| def render_sidebar(): | |
| with st.sidebar: | |
| st.title("๐ ENA Chatbot PRO") | |
| st.caption("v3.0 | Optimized Engine") | |
| st.divider() | |
| lang = st.radio("๐ Language / ุงููุบุฉ", ["ุงูุนุฑุจูุฉ", "Franรงais"], horizontal=True) | |
| st.session_state.lang = "ar" if lang == "ุงูุนุฑุจูุฉ" else "fr" | |
| st.divider() | |
| show_conf = st.toggle("๐ ู ุคุดุฑ ุงูุซูุฉ", value=True) | |
| show_obs = st.toggle("๐ก Observability", value=False) | |
| if show_obs and "logs" in st.session_state and st.session_state.logs: | |
| log = st.session_state.logs[-1] | |
| st.metric("Latency", format_latency(log.get("total_ms", 0))) | |
| with st.expander("Performance Details"): | |
| st.json(log) | |
| st.divider() | |
| if st.button("๐๏ธ ู ุณุญ ุงูู ุญุงุฏุซุฉ", use_container_width=True): | |
| st.session_state.chat_messages = [] | |
| st.session_state.last_sources = [] | |
| st.rerun() | |
| # PDF Section | |
| st.divider() | |
| st.markdown("### ๐ ู ุณุชุดุงุฑ ุงููุซุงุฆู (PDF)") | |
| uploaded_pdf = st.file_uploader("ุงุฑูุน ูุซููุฉ (ุจูุงุบ/ุฑุงุฆุฏ ุฑุณู ู)", type=["pdf"]) | |
| if uploaded_pdf: | |
| with st.spinner("ุฌุงุฑู ูุฑุงุกุฉ ุงูู ูู..."): | |
| pdf_text = extract_pdf_text(uploaded_pdf.read()) | |
| st.session_state.uploaded_pdf_text = pdf_text | |
| st.success("โ ุชู ุช ูุฑุงุกุฉ ุงูู ูู! ุงุณุฃููู ูู ุตุงุฏุฑู ุณุชุดู ู ูุฐุง ุงูู ูู.") | |
| else: | |
| st.session_state.uploaded_pdf_text = "" | |
| st.divider() | |
| # DB Status | |
| try: | |
| db_count = 0 | |
| try: | |
| db_count = engine.vectordb._collection.count() | |
| except: | |
| # If collection get fails, we treat it as empty | |
| db_count = 0 | |
| if db_count > 0: | |
| st.success(f"โ ุงูู ุชููุฑ: {db_count} ู ุนููู ุฉ") | |
| else: | |
| st.warning("โ ๏ธ ูุงุนุฏุฉ ุงูุจูุงูุงุช ูุงุฑุบุฉ!") | |
| if st.button("๐๏ธ ุจูุงุก ุงููุงุนุฏุฉ ู ู ุงูู ูู ุงูุฃุตูู (JSON)"): | |
| with st.spinner("โจ ุฌุงุฑู ุงูู ุนุงูุฌุฉ... ูุฑุฌู ุงูุงูุชุธุงุฑ ุฏูููุชูู"): | |
| try: | |
| import build_chroma | |
| build_chroma.build() | |
| st.cache_resource.clear() | |
| st.success("โ ุชู ุงูุจูุงุก ุจูุฌุงุญ!") | |
| st.rerun() | |
| except Exception as e: | |
| st.error(f"โ ูุดู ุงูุจูุงุก: {e}") | |
| if st.button("๐ฅ ูู ุถุบุท ZIP (ุจุฏูู ุณุฑูุน)"): | |
| import zipfile | |
| if os.path.exists("chroma_ena_db.zip"): | |
| with zipfile.ZipFile("chroma_ena_db.zip", 'r') as z: | |
| z.extractall("./temp_db") | |
| src = "./temp_db/chroma_ena_db" if os.path.exists("./temp_db/chroma_ena_db") else "./temp_db" | |
| if not os.path.exists(CHROMA_PATH): os.makedirs(CHROMA_PATH) | |
| import shutil | |
| for f in os.listdir(src): | |
| shutil.move(os.path.join(src, f), os.path.join(CHROMA_PATH, f)) | |
| st.cache_resource.clear() | |
| st.success("โ ุชู ูู ุงูุถุบุท!") | |
| st.rerun() | |
| except Exception as e: | |
| st.error(f"โ ุฎุทุฃ ูู ุงูุงุชุตุงู") | |
| with st.expander("ุชูุงุตูู ุงูุฎุทุฃ ุงูุชููู"): | |
| st.code(str(e)) | |
| if st.button("๐งน ู ุณุญ ูุฅุนุงุฏุฉ ุจูุงุก"): | |
| import shutil | |
| if os.path.exists(CHROMA_PATH): shutil.rmtree(CHROMA_PATH) | |
| st.rerun() | |
| return show_conf | |
| def render_welcome(): | |
| is_ar = st.session_state.get("lang", "ar") == "ar" | |
| title = "๐ ENA Chatbot PRO" | |
| subtitle = "ู ุณุงุนุฏู ุงูุฐูู ููู ูุงุธุฑุงุช ูุงูุชูููู ุจุงูู ุฏุฑุณุฉ ุงููุทููุฉ ููุฅุฏุงุฑุฉ" if is_ar else "Votre assistant intelligent pour les concours et formations de l'ENA" | |
| st.markdown(f""" | |
| <div class="welcome-banner"> | |
| <h1>{title}</h1> | |
| <p>{subtitle}</p> | |
| </div> | |
| """, unsafe_allow_html=True) | |
| sug_label = "๐ก ุฃุณุฆูุฉ ู ูุชุฑุญุฉ:" if is_ar else "๐ก Questions suggรฉrรฉes :" | |
| st.markdown(f'<div style="text-align: {"right" if is_ar else "left"}; margin-top: 20px;">{sug_label}</div>', unsafe_allow_html=True) | |
| cols = st.columns(4) | |
| ar_questions = ["ุดุฑูุท ู ูุงุธุฑุฉ ุงูู ุฑุญูุฉ ุงูุนููุง", "ุงููุซุงุฆู ุงูู ุทููุจุฉ ููุชุณุฌูู", "ุจุฑูุงู ุฌ ู ูุงุธุฑุฉ ุฃ2", "ุชุงุฑูุฎ ูุชุญ ุงูู ูุงุธุฑุงุช ุงููุงุฏู ุฉ"] | |
| fr_questions = ["Conditions Cycle Supรฉrieur", "Documents d'inscription", "Programme Concours A2", "Dates des prochains concours"] | |
| questions = ar_questions if is_ar else fr_questions | |
| for i, q in enumerate(questions): | |
| if cols[i].button(q, key=f"q_{i}", use_container_width=True): | |
| st.session_state.pending_q = q | |
| st.rerun() | |
| # โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ | |
| # ๐ MAIN APP | |
| # โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ | |
| def main(): | |
| show_conf = render_sidebar() | |
| if "chat_messages" not in st.session_state: | |
| st.session_state.chat_messages = [] | |
| if not st.session_state.chat_messages: | |
| render_welcome() | |
| # Layout for Chat + Sources | |
| col_chat, col_sources = st.columns([0.7, 0.3]) | |
| with col_chat: | |
| # Display history | |
| for msg in st.session_state.chat_messages: | |
| with st.chat_message(msg["role"]): | |
| st.markdown(msg["content"]) | |
| # โโ ุนุฑุถ ุงูุงูุชุฑุงุญุงุช ุจุนุฏ ุขุฎุฑ ุฅุฌุงุจุฉ โโ | |
| suggestions = st.session_state.get("last_suggestions", []) | |
| if suggestions and st.session_state.chat_messages: | |
| lang = st.session_state.get("lang", "ar") | |
| sug_label = "๐ก ูุฏ ููู ู ุฃูุถุงู:" if lang == "ar" else "๐ก Vous pourriez aussi demander:" | |
| st.markdown(f"---\n**{sug_label}**") | |
| sug_cols = st.columns(len(suggestions)) | |
| for i, sug in enumerate(suggestions): | |
| # ุฏุนู ุงููุณุฎุชูู: dict (ุฌุฏูุฏ) ุฃู string (ูุฏูู ) | |
| if isinstance(sug, dict): | |
| icon = sug.get("icon", "๐ก") | |
| title = sug.get("title", "") | |
| desc = sug.get("desc", "") | |
| question = sug.get("question", title) | |
| label = f"{icon} **{title}**\n_{desc}_" | |
| else: | |
| label = sug | |
| question = sug | |
| with sug_cols[i]: | |
| if isinstance(sug, dict): | |
| st.markdown(f"{icon} **{title}**") | |
| st.caption(desc) | |
| if st.button("ุงุณุฃู โ" if lang == "ar" else "Demander โ", key=f"sug_{len(st.session_state.chat_messages)}_{i}"): | |
| st.session_state.last_suggestions = [] | |
| st.session_state.pending_q = question | |
| st.rerun() | |
| else: | |
| if st.button(sug, key=f"sug_{len(st.session_state.chat_messages)}_{i}"): | |
| st.session_state.last_suggestions = [] | |
| st.session_state.pending_q = question | |
| st.rerun() | |
| # Input | |
| is_ar = st.session_state.get("lang", "ar") == "ar" | |
| placeholder = "ุงุณุฃู ุนู ุงูู ูุงุธุฑุงุชุ ุงูุดุฑูุทุ ุฃู ุฅุฌุฑุงุกุงุช ุงูุชูููู..." if is_ar else "Posez des questions sur les concours, conditions ou procรฉdures..." | |
| prompt = st.chat_input(placeholder) | |
| pending_q = st.session_state.pop("pending_q", None) | |
| final_q = prompt or pending_q | |
| if final_q: | |
| with st.chat_message("user"): | |
| st.markdown(final_q) | |
| with st.chat_message("assistant"): | |
| t_start = time.time() | |
| lang = detect_lang(final_q) | |
| history = st.session_state.chat_messages[:] | |
| # ูุถูู ุณุคุงู ุงูู ุณุชุฎุฏู ููุชุงุฑูุฎ ููุง ุจุนุฏ ู ุง ูุฃุฎุฐ ูุณุฎุฉ ู ู ุงูุชุงุฑูุฎ ุงููุฏูู | |
| st.session_state.chat_messages.append({"role": "user", "content": final_q}) | |
| # โโ ุฅุฐุง ุงูู ุณุชุฎุฏู ุฃุฌุงุจ ุนูู ุฃุณุฆูุฉ ุงูุชุฃูู โโ | |
| if st.session_state.get("waiting_for_profile"): | |
| st.session_state.waiting_for_profile = False | |
| with st.spinner("๐ ุฌุงุฑู ุชูููู ู ููู..."): | |
| evaluation = agent.evaluate_user_profile(final_q, lang) | |
| st.markdown(evaluation) | |
| st.session_state.chat_messages.append({"role": "assistant", "content": evaluation}) | |
| st.session_state.last_tool = "evaluate_profile" | |
| st.rerun() | |
| # โโ Agentic RAG โโ | |
| else: | |
| spinner_msg = "๐ค ุฌุงุฑู ุงูุชูููุฑ..." if lang == "ar" else "๐ค Recherche en cours..." | |
| with st.spinner(spinner_msg): | |
| uploaded_pdf = st.session_state.get("uploaded_pdf_text", "") | |
| result = agent.run(final_q, history, uploaded_pdf) | |
| full_response = result["answer"] | |
| tool_used = result.get("tool_used", "") | |
| # ุฅุฐุง ูุญุชุงุฌ ู ุนููู ุงุช ุงูู ุณุชุฎุฏู | |
| if result.get("needs_user_input"): | |
| st.session_state.waiting_for_profile = True | |
| st.markdown(full_response) | |
| # โโ ุงูุชุฑุงุญุงุช ุฐููุฉ โ ูุญูุธูุง ูู session_state ูุจู rerun โโ | |
| if tool_used not in ("chat", "ask_user_profile") and not result.get("needs_user_input"): | |
| with st.spinner("๐ก ..."): | |
| suggestions = agent.generate_suggestions(final_q, full_response, lang) | |
| st.session_state.last_suggestions = suggestions if suggestions else [] | |
| else: | |
| st.session_state.last_suggestions = [] | |
| st.session_state.chat_messages.append({"role": "assistant", "content": full_response}) | |
| st.session_state.last_tool = tool_used | |
| st.session_state.last_sources = result.get("sources", []) | |
| # Log | |
| st.session_state.logs = st.session_state.get("logs", []) | |
| st.session_state.logs.append({ | |
| "total_ms": round((time.time() - t_start) * 1000), | |
| "tool": tool_used | |
| }) | |
| st.rerun() | |
| with col_sources: | |
| is_ar = st.session_state.get("lang", "ar") == "ar" | |
| header = "๐ค ุงูุฃุฏุงุฉ ุงูู ุณุชุฎุฏู ุฉ" if is_ar else "๐ค Outil utilisรฉ" | |
| st.subheader(header) | |
| tool_used = st.session_state.get("last_tool", "") | |
| tool_labels = { | |
| "search_db": ("๐ ุจุญุซ ูู ูุงุนุฏุฉ ุงูุจูุงูุงุช", "๐ Recherche DB"), | |
| "get_concours_page": ("๐ ุตูุญุฉ ุงูู ูุงุธุฑุฉ ุงููุงู ูุฉ", "๐ Page complรจte du concours"), | |
| "ask_user_profile": ("๐ค ุชูููู ุงูู ูู ุงูุดุฎุตู", "๐ค รvaluation du profil"), | |
| "evaluate_profile": ("โ ุชูููู ุงูุชุฃูู", "โ รvaluation รฉligibilitรฉ"), | |
| "summarize_url": ("๐ ุชูุฎูุต ู ูุงู", "๐ Rรฉsumรฉ article"), | |
| "chat": ("๐ฌ ู ุญุงุฏุซุฉ ุนุงู ุฉ", "๐ฌ Conversation"), | |
| "search_db (fallback)": ("๐ ุจุญุซ (ุงุญุชูุงุทู)", "๐ Recherche (fallback)"), | |
| } | |
| if tool_used and tool_used in tool_labels: | |
| label = tool_labels[tool_used][0] if is_ar else tool_labels[tool_used][1] | |
| st.success(label) | |
| # ุดุฑุญ ููุด ุงุฎุชุงุฑ ูุงุฐู ุงูุฃุฏุงุฉ | |
| explanations = { | |
| "search_db": "ุณุคุงู ุนุงู โ ุจุญุซุช ูู ูุงุนุฏุฉ ุงูุจูุงูุงุช", | |
| "get_concours_page": "ุณุคุงู ุนู ุดุฑูุท/ูุซุงุฆู โ ุฌุจุช ุงูุตูุญุฉ ูุงู ูุฉ ู ุจุงุดุฑุฉ ู ู ุงูู ููุน", | |
| "ask_user_profile": "ุณุคุงู ุดุฎุตู โ ุทูุจุช ู ุนููู ุงุชู ูุชูููู ุชุฃููู", | |
| "evaluate_profile": "ูููู ุช ู ููู ุจูุงุกู ุนูู ุงููุซุงุฆู ุงูุฑุณู ูุฉ", | |
| "summarize_url": "ูุฎูุตุช ุงูู ูุงู ุงูู ุทููุจ", | |
| } | |
| if tool_used in explanations: | |
| st.caption(explanations[tool_used]) | |
| # ุนุฑุถ ุงูู ุตุงุฏุฑ | |
| sources = st.session_state.get("last_sources", []) | |
| if sources: | |
| st.markdown("---") | |
| src_header = "๐ ุงูู ุตุงุฏุฑ ุฐุงุช ุงูุตูุฉ" if is_ar else "๐ Sources liรฉes" | |
| st.subheader(src_header) | |
| for i, src in enumerate(sources): | |
| url = src.get("url", "") | |
| page_name = src.get("page_name", "") | |
| content = src.get("content", "") | |
| if url: | |
| with st.expander(f"ุงูู ุตุฏุฑ [{i+1}]: {page_name}"): | |
| st.markdown(f"[{url}]({url})") | |
| if content: | |
| st.text_area("ุงููุต ุงูู ุณุชุฎุฑุฌ", content, height=120, disabled=True, key=f"src_{i}_{len(st.session_state.chat_messages)}") | |
| else: | |
| info_msg = "ุณุชุธูุฑ ุงูุฃุฏุงุฉ ุงูู ุณุชุฎุฏู ุฉ ูุงูู ุตุงุฏุฑ ููุง" if is_ar else "L'outil utilisรฉ et les sources apparaรฎtront ici" | |
| st.info(info_msg) | |
| if __name__ == "__main__": | |
| main() |