ENA-Chatbot / app.py
Ines1994's picture
Upload 3 files
a1b2408 verified
Raw History Blame Contribute Delete
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)
@st.cache_resource
def get_engine():
token = os.getenv("GROQ_TOKEN") or st.secrets.get("GROQ_TOKEN")
return ENAEngine(groq_token=token)
@st.cache_resource
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()