""" GENAURA - Copilote post-appel La Poste Application Streamlit, prototype hackathon 2026. Mécaniques techniques (à connaître pour le pitch jury) : Tchat assistant : LLM + RAG sur Corpus A (documents officiels). Génération du résumé : LLM seul (reformulation des champs saisis). Génération du mail : LLM + RAG sur Corpus A. Fiche client unifiée : requête filtrée sur Corpus B (= archives), pas de RAG. Tableau de bord : agrégations statistiques sur Corpus B, pas de RAG. Frontière étanche : Corpus B (données client) n'alimente JAMAIS le tchat ni la génération. """ import streamlit as st import ollama import chromadb from chromadb.utils import embedding_functions import json import os import re import tempfile from datetime import datetime from generate_pdf import generer_pdf_resume # Traduction via Google Translate (deep-translator). # Fallback LLM si la librairie ou le réseau ne sont pas disponibles. try: from deep_translator import GoogleTranslator GOOGLE_TRANSLATE_DISPONIBLE = True except ImportError: GOOGLE_TRANSLATE_DISPONIBLE = False # Transcription audio via Google Speech Recognition. # Fallback simulé si la librairie n'est pas disponible. try: import speech_recognition as sr SPEECH_RECOGNITION_DISPONIBLE = True except ImportError: SPEECH_RECOGNITION_DISPONIBLE = False # === CONFIGURATION === MODELE_LLM = "mistral" MODELE_EMBED = "nomic-embed-text" NB_RESULTATS_RAG = 3 TYPES_DEMANDE = [ "Réclamation", "Information", "Suivi de colis", "Modification de contrat", "Remboursement", "Escalade", "Autre" ] CANAUX_SUIVI = ["E-mail", "Courrier", "Rappel téléphonique"] CHEMIN_ARCHIVES = "./archives" CHEMIN_DATA = "./data" CHEMIN_PROMPTS = "./prompts" # === CHARGEMENT DES DONNÉES STATIQUES === def charger_langues(): """Charge la configuration des langues supportées (pas de cache, on relit à chaque fois).""" chemin = os.path.join(CHEMIN_DATA, "langues.json") with open(chemin, "r", encoding="utf-8") as f: return json.load(f) @st.cache_data def charger_historique_clients(): """Charge l'historique mock des clients (pour démo de la fiche client unifiée).""" chemin = os.path.join(CHEMIN_DATA, "historique_clients.json") if not os.path.exists(chemin): return {} with open(chemin, "r", encoding="utf-8") as f: return json.load(f) @st.cache_data def charger_exemples_demo(): """Charge les 5 exemples démo (un par typologie).""" chemin = os.path.join(CHEMIN_DATA, "exemples_demo.json") if not os.path.exists(chemin): return {} with open(chemin, "r", encoding="utf-8") as f: return json.load(f) # === INITIALISATION RAG (Corpus A) === @st.cache_resource def init_chroma(): """Connexion à ChromaDB (Corpus A : documents officiels La Poste).""" client = chromadb.PersistentClient(path="./chroma_db") ollama_ef = embedding_functions.OllamaEmbeddingFunction( url="http://localhost:11434/api/embeddings", model_name=MODELE_EMBED ) collection = client.get_or_create_collection( name="docs_la_poste", embedding_function=ollama_ef ) return collection def charger_prompt(nom_fichier: str) -> str: """Charge un prompt depuis le dossier prompts/.""" chemin = os.path.join(CHEMIN_PROMPTS, nom_fichier) with open(chemin, "r", encoding="utf-8") as f: return f.read() def rechercher_contexte(collection, texte_requete: str, n_results: int = NB_RESULTATS_RAG) -> str: """Recherche vectorielle dans Corpus A (RAG).""" results = collection.query( query_texts=[texte_requete], n_results=n_results ) if results["documents"] and results["documents"][0]: return "\n\n---\n\n".join(results["documents"][0]) return "(Aucun document de référence trouvé.)" def generer_avec_ollama(prompt: str, temperature: float = 0.3) -> str: """Appel LLM Ollama.""" response = ollama.chat( model=MODELE_LLM, messages=[{"role": "user", "content": prompt}], options={"temperature": temperature} ) return response["message"]["content"] def traduire_texte(texte_fr: str, code_google: str, nom_langue: str) -> tuple: """ Traduit un texte français vers la langue cible. Stratégie en 2 niveaux : 1. Google Translate via deep-translator (rapide et fiable, y compris arabe). 2. Fallback LLM Mistral si réseau coupé ou librairie absente. Retourne (texte_traduit, methode_utilisee) où methode est 'google', 'llm' ou 'erreur'. """ # Tentative 1, Google Translate if GOOGLE_TRANSLATE_DISPONIBLE: try: # Google Translate accepte ~5000 caractères par appel. # Pour un mail post-appel c'est largement suffisant. traducteur = GoogleTranslator(source="fr", target=code_google) texte_traduit = traducteur.translate(texte_fr) if texte_traduit and texte_traduit.strip(): return texte_traduit, "google" except Exception as e: st.warning( f"Google Translate indisponible ({type(e).__name__}). " "Bascule sur traduction LLM (qualité variable selon la langue)." ) # Tentative 2, fallback LLM try: prompt_trad = charger_prompt("prompt_traduction.txt").format( langue_cible=nom_langue, mail_francais=texte_fr ) texte_traduit = generer_avec_ollama(prompt_trad, temperature=0.2) return texte_traduit, "llm" except Exception: return f"(Traduction {nom_langue} indisponible. Le mail français est utilisable seul.)", "erreur" def generer_paragraphe_aide_fr(email_fr: str, donnees_appel: dict, mode_falc: bool = False) -> str: """ Génère un seul paragraphe court en français, destiné à être traduit dans la langue d'origine du demandeur. Règle métier : - le mail administratif officiel reste en français ; - le paragraphe traduit sert uniquement d'aide à la compréhension ; - aucune information absente du mail français ne doit être ajoutée. """ consigne_falc = ( "Utilise des phrases très courtes, un vocabulaire courant et une structure simple." if mode_falc else "Utilise un français clair, professionnel et accessible." ) prompt = f""" À partir du mail administratif français ci-dessous, rédige un seul paragraphe court en français. Ce paragraphe sera traduit dans la langue d'origine du demandeur. Contraintes obligatoires : - 4 phrases maximum. - Ne pas saluer. - Ne pas signer. - Ne pas mettre de titre. - Ne pas ajouter d'information absente du mail français. - Mentionner que la réponse officielle est le texte français au-dessus. - Résumer uniquement l'objet de la réponse, l'action prévue, la démarche attendue et les délais si présents. - {consigne_falc} Contexte d'appel : Client : {donnees_appel.get('nom_client', '')} Référence client : {donnees_appel.get('ref_client', '')} Objet : {donnees_appel.get('objet_appel', '')} Type de demande : {donnees_appel.get('type_demande', '')} Mail administratif français : {email_fr} """ try: paragraphe = generer_avec_ollama(prompt, temperature=0.2).strip() if paragraphe: # Sécurité d'affichage : un paragraphe, pas de markdown lourd. paragraphe = paragraphe.replace("###", "").replace("##", "").replace("#", "") paragraphe = " ".join(ligne.strip() for ligne in paragraphe.splitlines() if ligne.strip()) return paragraphe except Exception: pass objet = donnees_appel.get("objet_appel", "votre demande") return ( "La réponse officielle de l'administration est le texte français au-dessus. " f"Elle concerne {objet}. " "Les suites données à votre demande sont indiquées dans cette réponse. " "En cas de doute, vous pouvez contacter le conseiller indiqué dans le mail." ) # === SURLIGNAGE DES MOTS-CLÉS === # Patterns regex et couleurs associées PATTERNS_SURLIGNAGE = [ # Marqueur [À VÉRIFIER] : fond jaune, priorité maximale (r'\[À VÉRIFIER[^\]]*\]|\[À VERIFIER[^\]]*\]|\[A VERIFIER[^\]]*\]', 'background-color: #FFEB3B; color: #000; padding: 1px 4px; border-radius: 3px; font-weight: bold;'), # Dates au format DD/MM/YYYY ou DD/MM/YY (r'\b\d{1,2}/\d{1,2}/\d{2,4}\b', 'color: #C62828; font-weight: bold;'), # Heures et durées (48h, 5 jours, 10 jours ouvrés, 2 mois) (r'\b\d+\s*(heures?|h\b|jours?\s*(ouvrés?|ouvrables?)?|j\b|mois|semaines?|minutes?|min\b)\b', 'color: #E65100; font-weight: bold;'), # Montants en euros (r'\b\d+([\s,.]\d+)?\s*(EUR|euros?|€)\b', 'color: #2E7D32; font-weight: bold;'), # Références La Poste (CL-2026-00451, APP-20260502-100810, etc.) (r'\b[A-Z]{2,4}-\d{2,}(-\d+)?\b', 'color: #1565C0; font-weight: bold;'), # Numéros de suivi colis (ex Colissimo 6A12345678901, ou EZ123456789FR) (r'\b\d?[A-Z]\d{10,}\b|\b[A-Z]{2}\d{9,}[A-Z]{0,2}\b', 'color: #1565C0; font-weight: bold;'), # Mots-clés d'engagement (verbes d'action) (r'\b(enquête|rappel|envoi|remboursement|confirmation|résiliation|renvoi|escalade|réclamation)\b', 'color: #6A1B9A; font-style: italic;'), ] def surligner_mots_cles(texte: str) -> str: """ Surligne les mots-clés importants dans le texte. Retourne du HTML utilisable avec st.markdown(unsafe_allow_html=True). """ if not texte: return "" # Échapper les caractères HTML dangereux texte_html = ( texte.replace("&", "&") .replace("<", "<") .replace(">", ">") ) # Conserver les sauts de ligne en HTML texte_html = texte_html.replace("\n", "
") # Appliquer les patterns un par un for pattern, style in PATTERNS_SURLIGNAGE: def remplacer(match): return f'{match.group(0)}' texte_html = re.sub(pattern, remplacer, texte_html, flags=re.IGNORECASE) return texte_html def afficher_apercu_surligne(texte: str, titre: str = "Aperçu coloré (mots-clés surlignés)"): """Affiche un texte avec mots-clés surlignés dans un encadré.""" if not texte: return html = surligner_mots_cles(texte) bloc = f"""
{html}
""" st.markdown(f"**{titre}**") st.markdown(bloc, unsafe_allow_html=True) st.caption( "Légende : " "🟡 [À VÉRIFIER] | " "🔴 dates | " "🟠 délais | " "🟢 montants | " "🔵 références | " "🟣 actions" ) # === TRANSCRIPTION AUDIO === def transcrire_audio(audio_bytes: bytes, langue: str = "fr-FR") -> tuple: """ Transcrit un fichier audio (bytes) en texte via Google Speech Recognition. Retourne (texte_transcrit, methode_utilisee). Méthodes possibles : 'google', 'erreur'. """ if not SPEECH_RECOGNITION_DISPONIBLE: return ( "(Transcription indisponible : SpeechRecognition non installé. " "Lance `pip install SpeechRecognition` puis relance Streamlit.)", "erreur" ) try: # Sauvegarde temporaire de l'audio with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f: f.write(audio_bytes) chemin_audio = f.name try: recognizer = sr.Recognizer() with sr.AudioFile(chemin_audio) as source: audio_data = recognizer.record(source) texte = recognizer.recognize_google(audio_data, language=langue) return texte, "google" finally: try: os.unlink(chemin_audio) except OSError: pass except sr.UnknownValueError: return ("(Audio inaudible ou non reconnu. Réessaie en parlant plus distinctement.)", "erreur") except sr.RequestError as e: return (f"(Service Google indisponible : {e}. Vérifie ta connexion internet.)", "erreur") except Exception as e: return (f"(Erreur de transcription : {type(e).__name__}.)", "erreur") # === ARCHIVAGE (Corpus B = archives) === def sauvegarder_archive(donnees: dict) -> str: """Sauvegarde dans le dossier archives/ (Corpus B).""" os.makedirs(CHEMIN_ARCHIVES, exist_ok=True) chemin = os.path.join(CHEMIN_ARCHIVES, f"{donnees['id_appel']}.json") with open(chemin, "w", encoding="utf-8") as f: json.dump(donnees, f, ensure_ascii=False, indent=2) return chemin def lire_archives() -> list: """Lit toutes les archives (Corpus B).""" if not os.path.exists(CHEMIN_ARCHIVES): return [] archives = [] for filename in sorted(os.listdir(CHEMIN_ARCHIVES), reverse=True): if filename.endswith(".json"): with open(os.path.join(CHEMIN_ARCHIVES, filename), "r", encoding="utf-8") as f: archives.append(json.load(f)) return archives # === FICHE CLIENT UNIFIÉE (requête filtrée sur Corpus B) === def chercher_appels_client(ref_client: str) -> dict: """ Requête filtrée sur Corpus B (mock + archives réelles). Retourne les infos client + historique d'appels. Pas de RAG, pas de LLM : juste un filtre par référence. """ historique_mock = charger_historique_clients() archives_reelles = lire_archives() # Données mock du client info_client = historique_mock.get(ref_client, {}) appels = list(info_client.get("appels_passes", [])) # Ajout des archives réelles correspondant à ce client for archive in archives_reelles: if archive.get("ref_client") == ref_client: appels.append({ "date": archive.get("date", "")[:10], "id_appel": archive.get("id_appel", ""), "conseiller": archive.get("nom_conseiller", ""), "type_demande": archive.get("type_demande", ""), "objet": archive.get("objet_appel", ""), "resume_court": archive.get("resume_valide", "")[:200] + "...", "statut": archive.get("statut", "") }) # Tri par date décroissante appels.sort(key=lambda x: x.get("date", ""), reverse=True) return { "info_client": info_client, "appels": appels } # === TABLEAU DE BORD CONSEILLER (agrégations sur Corpus B) === def calculer_stats_conseiller(nom_conseiller: str) -> dict: """ Agrégations sur Corpus B filtrées sur le conseiller connecté. Pas de RAG : juste des comptages et des moyennes. """ historique_mock = charger_historique_clients() archives_reelles = lire_archives() # Combinaison mock + archives pour avoir des stats visibles dès le départ tous_appels = [] for client in historique_mock.values(): for appel in client.get("appels_passes", []): if appel.get("conseiller") == nom_conseiller: tous_appels.append(appel) for archive in archives_reelles: if archive.get("nom_conseiller") == nom_conseiller: tous_appels.append({ "date": archive.get("date", "")[:10], "type_demande": archive.get("type_demande", ""), "modifie_par_conseiller": archive.get("modifie_par_conseiller", False), "modifie_avant_envoi": archive.get("modifie_avant_envoi", False) }) nb_appels = len(tous_appels) # Comptage par type de demande types_demande = {} for appel in tous_appels: t = appel.get("type_demande", "Autre") types_demande[t] = types_demande.get(t, 0) + 1 # Taux de modification nb_archives_conseiller = sum( 1 for a in archives_reelles if a.get("nom_conseiller") == nom_conseiller ) nb_modif_resume = sum( 1 for a in archives_reelles if a.get("nom_conseiller") == nom_conseiller and a.get("modifie_par_conseiller") ) nb_modif_mail = sum( 1 for a in archives_reelles if a.get("nom_conseiller") == nom_conseiller and a.get("modifie_avant_envoi") ) taux_modif_resume = ( round(100 * nb_modif_resume / nb_archives_conseiller) if nb_archives_conseiller else 0 ) taux_modif_mail = ( round(100 * nb_modif_mail / nb_archives_conseiller) if nb_archives_conseiller else 0 ) return { "nb_appels": nb_appels, "types_demande": types_demande, "taux_modif_resume": taux_modif_resume, "taux_modif_mail": taux_modif_mail, "nb_archives_reelles": nb_archives_conseiller } # === SIDEBAR : CHAT ASSISTANT RAG (Corpus A) === QUESTIONS_ASSISTANT_RAG = [ "Quel est le délai de remboursement d’un Colissimo ?", "Que faire si un colis est déclaré perdu ?", "Quand faut-il escalader une réclamation ?", "Comment expliquer le suivi d’un colis au client ?", "Quelles informations demander avant une modification de contrat ?", "Quel ton utiliser dans une réponse client La Poste ?", ] def traiter_question_assistant(question: str): """ Traite une question du tchat assistant avec RAG sur Corpus A uniquement. Aucune donnée client ni archive Corpus B n'est utilisée. """ if not question or not question.strip(): return question = question.strip() with st.spinner("Recherche dans les documents La Poste..."): collection = init_chroma() contexte = rechercher_contexte(collection, question) prompt_template = charger_prompt("prompt_chat.txt") prompt = prompt_template.format( contexte_rag=contexte, question=question ) reponse = generer_avec_ollama(prompt, temperature=0.2) st.session_state.chat_messages.append({ "role": "user", "content": question }) st.session_state.chat_messages.append({ "role": "assistant", "content": reponse, "sources": contexte }) st.rerun() def afficher_chat_assistant(): """ Chat assistant en sidebar. Mécanique : LLM + RAG sur Corpus A uniquement. Aucune donnée client n'est utilisée. """ st.sidebar.markdown("### Assistant La Poste (RAG)") st.sidebar.caption( "Posez vos questions sur la documentation La Poste pendant l'appel. " "Source : Corpus A (procédures, FAQ, conditions)." ) if "chat_messages" not in st.session_state: st.session_state.chat_messages = [] # Affichage de l'historique for msg in st.session_state.chat_messages: if msg["role"] == "user": st.sidebar.markdown(f"**Vous :** {msg['content']}") else: st.sidebar.markdown(f"**Assistant :** {msg['content']}") if msg.get("sources"): with st.sidebar.expander("Sources documentaires"): st.text(msg["sources"]) st.sidebar.markdown("---") # Input libre question = st.sidebar.text_area( "Votre question", key="chat_input", height=80, placeholder="Ex : Quel est le délai de remboursement Colissimo ?" ) col_a, col_b = st.sidebar.columns(2) with col_a: if st.button("Poser", key="chat_send", use_container_width=True): traiter_question_assistant(question) with col_b: if st.button("Effacer", key="chat_clear", use_container_width=True): st.session_state.chat_messages = [] st.rerun() # Questions prêtes à l'emploi, alignées sur les documents indexés du Corpus A. st.sidebar.markdown("**Questions prêtes**") st.sidebar.caption( "Cliquez sur une question pour interroger directement les documents La Poste." ) for i, question_suggeree in enumerate(QUESTIONS_ASSISTANT_RAG): if st.sidebar.button( question_suggeree, key=f"question_rag_{i}", use_container_width=True ): traiter_question_assistant(question_suggeree) # === ONGLET 1 : NOUVEL APPEL (les 5 étapes) === def afficher_onglet_nouvel_appel(): """Flux principal en 5 étapes.""" # Initialisation if "etape" not in st.session_state: st.session_state.etape = 1 if "donnees_appel" not in st.session_state: st.session_state.donnees_appel = {} # Barre de progression etapes_labels = [ "1. Saisie", "2. Résumé IA", "3. Validation résumé", "4. E-mail", "5. Archivé" ] cols = st.columns(5) for i, label in enumerate(etapes_labels): with cols[i]: if i + 1 < st.session_state.etape: st.success(label) elif i + 1 == st.session_state.etape: st.info(label) else: st.caption(label) st.divider() # Aiguillage par étape if st.session_state.etape == 1: etape_1_saisie() elif st.session_state.etape == 2: etape_2_resume() elif st.session_state.etape == 3: etape_3_validation() elif st.session_state.etape == 4: etape_4_mail() elif st.session_state.etape == 5: etape_5_archivage() def etape_1_saisie(): """Étape 1 : saisie des informations.""" st.subheader("Saisie des informations de l'appel") # Boutons d'exemples démo (5 sur la même ligne) exemples = charger_exemples_demo() if exemples: st.markdown("**Mes personas - Charger un exemple démo :**") cols_demo = st.columns(len(exemples)) for i, (cle, contenu) in enumerate(exemples.items()): with cols_demo[i]: if st.button(contenu["libelle_bouton"], key=f"demo_{cle}", use_container_width=True): st.session_state.exemple = contenu["donnees"] st.rerun() st.divider() exemple = st.session_state.get("exemple", {}) # === Saisie vocale optionnelle === with st.expander("🎙️ Saisie vocale des notes (optionnel)", expanded=False): st.caption( "Enregistre tes notes oralement, l'IA les transcrira automatiquement. " "Pratique pour ne rien rater pendant l'appel." ) if not SPEECH_RECOGNITION_DISPONIBLE: st.warning( "SpeechRecognition non installé. " "Lance `pip install SpeechRecognition` puis relance Streamlit pour activer." ) else: audio_input = st.audio_input("Enregistrer", key="audio_recorder") if audio_input is not None: col_a1, col_a2 = st.columns([1, 3]) with col_a1: if st.button("Transcrire", key="btn_transcrire", type="primary"): with st.spinner("Transcription en cours..."): texte, methode = transcrire_audio(audio_input.getvalue()) if methode == "google": st.session_state["transcription_audio"] = texte st.success("Transcription terminée. Texte ajouté aux notes ci-dessous.") else: st.error(texte) st.session_state["transcription_audio"] = "" with col_a2: if st.session_state.get("transcription_audio"): st.markdown("**Transcription :**") st.info(st.session_state["transcription_audio"]) st.divider() with st.form("formulaire_appel"): col1, col2 = st.columns(2) with col1: nom_client = st.text_input("Nom du client", value=exemple.get("nom_client", "")) ref_client = st.text_input("Référence client", value=exemple.get("ref_client", "")) nom_conseiller = st.text_input( "Votre nom (conseiller)", value=exemple.get("nom_conseiller", "") ) with col2: type_demande = st.selectbox( "Type de demande", TYPES_DEMANDE, index=TYPES_DEMANDE.index(exemple.get("type_demande", "Réclamation")) if exemple.get("type_demande") in TYPES_DEMANDE else 0 ) canal_suivi = st.selectbox( "Canal de suivi prévu", CANAUX_SUIVI, index=CANAUX_SUIVI.index(exemple.get("canal_suivi", "E-mail")) ) objet_appel = st.text_input("Objet de l'appel", value=exemple.get("objet_appel", "")) # Notes : combinaison de la transcription audio (si présente) et de l'exemple valeur_notes_initiale = exemple.get("notes_appel", "") transcription = st.session_state.get("transcription_audio", "") if transcription and not valeur_notes_initiale: valeur_notes_initiale = transcription elif transcription and valeur_notes_initiale: valeur_notes_initiale = f"{valeur_notes_initiale}\n\n[Transcription vocale]\n{transcription}" notes_appel = st.text_area( "Notes de l'appel", value=valeur_notes_initiale, height=150 ) engagements = st.text_area( "Engagements pris", value=exemple.get("engagements", ""), height=100 ) elements_sensibles = st.text_area( "Éléments sensibles (optionnel)", value=exemple.get("elements_sensibles", ""), height=80 ) submit = st.form_submit_button("Générer le résumé IA", type="primary") if submit: if not all([nom_client, ref_client, nom_conseiller, objet_appel, notes_appel, engagements]): st.error("Merci de remplir les champs obligatoires.") else: st.session_state.donnees_appel = { "id_appel": f"APP-{datetime.now().strftime('%Y%m%d-%H%M%S')}", "date": datetime.now().isoformat(), "nom_client": nom_client, "ref_client": ref_client, "nom_conseiller": nom_conseiller, "type_demande": type_demande, "objet_appel": objet_appel, "notes_appel": notes_appel, "engagements": engagements, "elements_sensibles": elements_sensibles, "canal_suivi": canal_suivi, "statut": "en_cours_de_saisie" } st.session_state.etape = 2 st.rerun() # Affichage de la fiche client si déjà connue (Corpus B) if exemple.get("ref_client"): afficher_apercu_fiche_client(exemple["ref_client"]) def afficher_apercu_fiche_client(ref_client: str): """Aperçu rapide de la fiche client (continuité de service).""" donnees = chercher_appels_client(ref_client) if donnees["appels"]: with st.expander(f"Historique de {ref_client} ({len(donnees['appels'])} appel(s) précédent(s))"): for appel in donnees["appels"][:3]: st.markdown( f"**{appel.get('date', '')}** " f"({appel.get('type_demande', '')}) " f"par {appel.get('conseiller', '')} : " f"{appel.get('objet', '')}" ) st.caption(appel.get("resume_court", "")) def etape_2_resume(): """Étape 2 : génération du résumé interne (LLM seul, pas de RAG).""" st.subheader("Résumé de l'appel généré par l'IA") d = st.session_state.donnees_appel with st.spinner("Génération du résumé en cours (LLM seul, pas de RAG)..."): prompt_template = charger_prompt("prompt_resume.txt") prompt = prompt_template.format( nom_client=d["nom_client"], ref_client=d["ref_client"], type_demande=d["type_demande"], objet_appel=d["objet_appel"], notes_appel=d["notes_appel"], engagements=d["engagements"], elements_sensibles=d["elements_sensibles"] or "(Aucun)", canal_suivi=d["canal_suivi"], date=datetime.now().strftime("%d/%m/%Y à %H:%M") ) resume_ia = generer_avec_ollama(prompt) st.session_state.donnees_appel["resume_ia"] = resume_ia st.session_state.donnees_appel["statut"] = "resume_genere" st.session_state.etape = 3 st.rerun() def etape_3_validation(): """Étape 3 : validation du résumé par le conseiller.""" st.subheader("Validation du résumé interne") d = st.session_state.donnees_appel st.caption(f"Appel {d['id_appel']} - {d['nom_client']}") # Affichage du contexte RAG anticipé (Corpus A) st.markdown("**Documents La Poste utilisés pour la suite (RAG, Corpus A) :**") with st.expander("Voir le contexte documentaire qui sera utilisé pour générer le mail"): collection = init_chroma() contexte_anticipe = rechercher_contexte(collection, d.get("resume_ia", "")) st.text(contexte_anticipe) st.session_state.donnees_appel["contexte_rag"] = contexte_anticipe st.markdown("Relisez et corrigez si besoin avant de valider.") resume_editable = st.text_area( "Résumé interne (jamais envoyé au client)", value=d.get("resume_ia", ""), height=350, key="resume_edit" ) # Aperçu coloré avec mots-clés surlignés afficher_apercu_surligne(resume_editable, titre="Aperçu coloré du résumé") col1, col2 = st.columns(2) with col1: if st.button("Valider le résumé", type="primary"): st.session_state.donnees_appel["resume_valide"] = resume_editable st.session_state.donnees_appel["modifie_par_conseiller"] = ( resume_editable != d.get("resume_ia", "") ) st.session_state.donnees_appel["statut"] = "resume_valide" st.session_state.etape = 4 st.rerun() with col2: if st.button("Régénérer le résumé"): st.session_state.etape = 2 st.rerun() def etape_4_mail(): """Étape 4 : génération du mail (LLM + RAG sur Corpus A) avec langue et FALC.""" st.subheader("E-mail client généré par l'IA") d = st.session_state.donnees_appel st.caption(f"Appel {d['id_appel']} - {d['nom_client']}") # === Paramètres de génération === langues = charger_langues() codes_langues = list(langues.keys()) libelles_langues = [langues[c]["libelle"] for c in codes_langues] # Détection langue préférée du client si présente dans le mock historique = charger_historique_clients() langue_pref_client = historique.get(d["ref_client"], {}).get("langue_preferee", "FR") index_defaut = codes_langues.index(langue_pref_client) if langue_pref_client in codes_langues else 0 st.markdown("**Paramètres du mail**") col_p1, col_p2 = st.columns(2) with col_p1: langue_choisie = st.selectbox( "Langue du mail", codes_langues, format_func=lambda c: langues[c]["libelle"], index=index_defaut, key="select_langue", help="Si autre que français, le mail reste en français et ajoute un paragraphe d'aide dans la langue choisie." ) with col_p2: mode_falc = st.toggle( "Mode FALC (clients seniors ou en illectronisme)", value=False, key="toggle_falc", help="Facile à Lire et à Comprendre : phrases courtes, vocabulaire simple, structure très claire." ) # Règle métier : le mail administratif officiel reste en français. # Si le demandeur a une langue différente du français, on ajoute dans le même mail # un paragraphe court d'aide à la compréhension dans la langue choisie. # On ne dépend pas du champ JSON "bilingue", afin d'éviter qu'une mauvaise configuration # bloque l'ajout du paragraphe traduit. config_langue = langues[langue_choisie] est_bilingue = langue_choisie != "FR" # Bandeau de debug pour diagnostiquer rapidement with st.expander("Détails techniques (debug)", expanded=False): st.markdown(f"- Langue choisie : **{langue_choisie}** ({config_langue.get('nom_natif', '?')})") st.markdown(f"- Paragraphe d'aide multilingue : **{'Oui' if est_bilingue else 'Non'}**") st.markdown( f"- Google Translate disponible : " f"**{'Oui' if GOOGLE_TRANSLATE_DISPONIBLE else 'Non, fallback LLM'}**" ) st.markdown(f"- Mode FALC : **{'Oui' if mode_falc else 'Non'}**") if not GOOGLE_TRANSLATE_DISPONIBLE: st.warning( "deep-translator non installé. Lance `pip install deep-translator` " "puis relance Streamlit pour activer la traduction propre." ) # Bouton régénération si paramètres changés parametres_actuels = (langue_choisie, mode_falc) # Détection automatique des changements de paramètres : on invalide le mail # si la langue ou le mode FALC ont changé depuis la dernière génération. parametres_precedents = d.get("parametres_mail") if parametres_precedents is not None and parametres_precedents != parametres_actuels: if "email_genere" in st.session_state.donnees_appel: del st.session_state.donnees_appel["email_genere"] if "methode_traduction" in st.session_state.donnees_appel: del st.session_state.donnees_appel["methode_traduction"] if "email_edit" in st.session_state: del st.session_state["email_edit"] st.info( "Paramètres modifiés. Régénération automatique du mail en cours." ) # Génération si pas encore fait (ou si paramètres ont changé) if "email_genere" not in d or st.session_state.get("force_regen_mail"): st.session_state["force_regen_mail"] = False with st.spinner("Recherche RAG dans le Corpus A..."): collection = init_chroma() contexte_rag = rechercher_contexte(collection, d["resume_valide"]) st.session_state.donnees_appel["contexte_rag"] = contexte_rag # Étape A, génération du mail français (toujours, même si bilingue) with st.spinner("Génération du mail français..."): instruction_niveau = ( "Mode FALC strict : phrases de 15 mots maximum, vocabulaire courant uniquement, " "pas de jargon La Poste, structure très claire en 3 parties (ce qui s'est passé, " "ce qu'on fait, ce qu'on attend de vous), une idée par phrase." if mode_falc else "Niveau de lecture standard, professionnel et accessible." ) prompt_email = charger_prompt("prompt_email.txt").format( nom_client=d["nom_client"], nom_conseiller=d["nom_conseiller"], resume_valide=d["resume_valide"], contexte_rag=contexte_rag, instruction_niveau=instruction_niveau ) email_fr = generer_avec_ollama(prompt_email) # Étape B, si la langue cible n'est pas le français, on ajoute seulement # un paragraphe d'aide dans la langue d'origine du demandeur. if est_bilingue: # Récupération du code Google Translate avec fallback sur le code en minuscules code_google = config_langue.get("code_google", langue_choisie.lower()) nom_langue = config_langue.get("nom_pour_traduction", config_langue.get("nom_natif", langue_choisie)) nom_natif = config_langue.get("nom_natif", langue_choisie) with st.spinner("Génération du paragraphe d'aide en français..."): paragraphe_aide_fr = generer_paragraphe_aide_fr( email_fr=email_fr, donnees_appel=d, mode_falc=mode_falc ) with st.spinner(f"Traduction du paragraphe d'aide vers {nom_langue}..."): paragraphe_traduit, methode = traduire_texte( texte_fr=paragraphe_aide_fr, code_google=code_google, nom_langue=nom_langue ) st.session_state.donnees_appel["methode_traduction"] = methode st.session_state.donnees_appel["paragraphe_aide_fr"] = paragraphe_aide_fr st.session_state.donnees_appel["paragraphe_aide_traduit"] = paragraphe_traduit # Un seul mail : réponse officielle en français, puis paragraphe d'aide traduit. email_genere = ( f"{email_fr.strip()}\n\n" f"---\n\n" f"Aide à la compréhension ({nom_natif})\n\n" f"{paragraphe_traduit.strip()}" ) else: email_genere = email_fr st.session_state.donnees_appel["email_genere"] = email_genere st.session_state["email_edit"] = email_genere st.session_state.donnees_appel["email_fr"] = email_fr st.session_state.donnees_appel["langue_mail"] = langue_choisie st.session_state.donnees_appel["mode_falc"] = mode_falc st.session_state.donnees_appel["parametres_mail"] = parametres_actuels st.session_state.donnees_appel["statut"] = "mail_genere" st.rerun() # Affichage et édition st.markdown("---") titre_mail = ( f"**Mail généré** ({langues[langue_choisie]['libelle']}" f"{', mode FALC actif' if mode_falc else ''})" ) st.markdown(titre_mail) methode = d.get("methode_traduction") if methode == "google": st.caption( "Paragraphe d'aide traduit par Google Translate. Marqueur [À VÉRIFIER] : éléments " "chiffrés ou contractuels à relire avant envoi." ) elif methode == "llm": st.caption( "Paragraphe d'aide traduit par LLM local (Google Translate indisponible). " "Vérifier la qualité avant envoi. Marqueur [À VÉRIFIER] sur les éléments sensibles." ) else: st.caption( "Marqueur [À VÉRIFIER] : éléments chiffrés ou contractuels à relire avant envoi." ) email_editable = st.text_area( "E-mail à envoyer au client", value=d.get("email_genere", ""), height=350, key="email_edit" ) # Aperçu coloré avec mots-clés surlignés (uniquement la partie française officielle si multilingue) if "---" in email_editable: partie_fr = email_editable.split("---", 1)[0] afficher_apercu_surligne(partie_fr, titre="Aperçu coloré du mail (version française)") else: afficher_apercu_surligne(email_editable, titre="Aperçu coloré du mail") # Aperçu client : réponse officielle en français + paragraphe d'aide traduit if est_bilingue and "---" in email_editable: with st.expander("Aperçu du mail client", expanded=False): parties = email_editable.split("---", 1) col_v1, col_v2 = st.columns(2) with col_v1: st.markdown("**Réponse officielle en français**") st.text(parties[0].strip()) with col_v2: st.markdown(f"**Paragraphe d'aide ({langues[langue_choisie]['nom_natif']})**") st.text(parties[1].strip() if len(parties) > 1 else "") # Pièce jointe inclure_pj = st.checkbox("Joindre le résumé de l'appel en PDF", value=True) if inclure_pj: with st.expander("Aperçu de la pièce jointe"): st.markdown(d.get("resume_valide", "")) st.markdown("---") col1, col2, col3 = st.columns(3) with col1: if st.button("Valider et envoyer", type="primary"): st.session_state.donnees_appel["email_envoye"] = email_editable st.session_state.donnees_appel["modifie_avant_envoi"] = ( email_editable != d.get("email_genere", "") ) st.session_state.donnees_appel["pj_resume_incluse"] = inclure_pj st.session_state.donnees_appel["date_envoi"] = datetime.now().isoformat() st.session_state.donnees_appel["statut"] = "envoye_et_archive" if inclure_pj: os.makedirs(CHEMIN_ARCHIVES, exist_ok=True) chemin_pdf = generer_pdf_resume( nom_client=d["nom_client"], ref_client=d["ref_client"], nom_conseiller=d["nom_conseiller"], resume_valide=d["resume_valide"], id_appel=d["id_appel"], email_envoye=email_editable, langue_mail=d.get("langue_mail", "FR") ) st.session_state.donnees_appel["chemin_pdf"] = chemin_pdf chemin_archive = sauvegarder_archive(st.session_state.donnees_appel) st.session_state.donnees_appel["chemin_archive"] = chemin_archive st.session_state.etape = 5 st.rerun() with col2: if st.button("Régénérer le mail"): st.session_state["force_regen_mail"] = True if "email_genere" in st.session_state.donnees_appel: del st.session_state.donnees_appel["email_genere"] st.rerun() with col3: if st.button("Retour au résumé"): if "email_genere" in st.session_state.donnees_appel: del st.session_state.donnees_appel["email_genere"] st.session_state.etape = 3 st.rerun() def etape_5_archivage(): """Étape 5 : récapitulatif et archivage.""" st.subheader("E-mail envoyé et appel archivé") d = st.session_state.donnees_appel st.success("E-mail envoyé (simulation).") st.success("Appel archivé dans le dossier client (Corpus B).") col1, col2 = st.columns(2) with col1: st.markdown("**Détails de l'appel**") st.markdown(f"- Référence : {d['id_appel']}") st.markdown(f"- Client : {d['nom_client']} (réf. {d['ref_client']})") st.markdown(f"- Type : {d['type_demande']}") st.markdown(f"- Conseiller : {d['nom_conseiller']}") st.markdown(f"- Langue mail : {d.get('langue_mail', 'FR')}") st.markdown(f"- Mode FALC : {'Oui' if d.get('mode_falc') else 'Non'}") with col2: st.markdown("**Traçabilité**") st.markdown(f"- Résumé modifié : {'Oui' if d.get('modifie_par_conseiller') else 'Non'}") st.markdown(f"- Mail modifié avant envoi : {'Oui' if d.get('modifie_avant_envoi') else 'Non'}") st.markdown(f"- PJ incluse : {'Oui' if d.get('pj_resume_incluse') else 'Non'}") with st.expander("Résumé interne validé"): st.markdown(d.get("resume_valide", "")) with st.expander("Mail envoyé"): st.markdown(d.get("email_envoye", "")) if d.get("chemin_pdf") and os.path.exists(d["chemin_pdf"]): with open(d["chemin_pdf"], "rb") as f: st.download_button( "Télécharger le PDF du résumé", data=f.read(), file_name=f"{d['id_appel']}_resume.pdf", mime="application/pdf" ) st.markdown("---") if st.button("Nouvel appel", type="primary"): st.session_state.etape = 1 st.session_state.donnees_appel = {} if "exemple" in st.session_state: del st.session_state["exemple"] st.rerun() # === ONGLET 2 : FICHE CLIENT UNIFIÉE (requête filtrée Corpus B) === def afficher_onglet_fiche_client(): st.subheader("Fiche client unifiée") st.caption( "Vue continuité de service : tous les appels passés du client courant. " "Source : Corpus B (archives). Mécanique : requête filtrée, pas de RAG." ) ref_input = st.text_input( "Référence client", value=st.session_state.get("donnees_appel", {}).get("ref_client", "CL-2026-00451"), help="Saisir la référence d'un client. Exemples mock : CL-2026-00451, CL-2026-00782." ) if ref_input: donnees = chercher_appels_client(ref_input) info = donnees["info_client"] appels = donnees["appels"] if not info and not appels: st.warning(f"Aucun client trouvé pour la référence {ref_input}.") return # Bloc info client st.markdown("### Informations client") col1, col2, col3 = st.columns(3) with col1: st.markdown(f"**Nom :** {info.get('nom_client', '(inconnu)')}") st.markdown(f"**Référence :** {ref_input}") with col2: st.markdown(f"**E-mail :** {info.get('email', '-')}") st.markdown(f"**Téléphone :** {info.get('telephone', '-')}") with col3: st.markdown(f"**Ville :** {info.get('ville', '-')}") st.markdown(f"**Langue préférée :** {info.get('langue_preferee', 'FR')}") # Bloc historique st.markdown(f"### Historique des appels ({len(appels)})") if not appels: st.info("Aucun appel passé pour ce client.") else: for appel in appels: with st.expander( f"{appel.get('date', '')} - " f"{appel.get('type_demande', '')} - " f"{appel.get('objet', '')}" ): st.markdown(f"**Conseiller :** {appel.get('conseiller', '')}") st.markdown(f"**Statut :** {appel.get('statut', '')}") st.markdown(f"**Résumé :** {appel.get('resume_court', '')}") # === ONGLET 3 : TABLEAU DE BORD CONSEILLER === def afficher_onglet_dashboard(): st.subheader("Tableau de bord personnel") st.caption( "Vos statistiques personnelles. " "Source : Corpus B filtré sur votre nom. Mécanique : agrégations, pas de RAG." ) nom = st.text_input( "Votre nom (conseiller)", value=st.session_state.get("donnees_appel", {}).get("nom_conseiller", "Sophie Martin") ) if not nom: return stats = calculer_stats_conseiller(nom) # KPIs col1, col2, col3 = st.columns(3) with col1: st.metric("Appels traités", stats["nb_appels"]) with col2: st.metric("Taux modification résumé", f"{stats['taux_modif_resume']}%") with col3: st.metric("Taux modification mail", f"{stats['taux_modif_mail']}%") st.caption( f"Dont {stats['nb_archives_reelles']} appel(s) issu(s) de cette session, " "le reste vient de l'historique simulé pour la démo." ) # Graphique types de demande st.markdown("### Répartition par type de demande") if stats["types_demande"]: st.bar_chart(stats["types_demande"]) else: st.info("Pas encore de données.") # Coach factuel st.markdown("### Coach personnel (factuel)") if stats["nb_appels"] == 0: st.info("Traitez un premier appel pour activer le coach.") else: if stats["taux_modif_mail"] > 60: st.warning( f"Vous modifiez {stats['taux_modif_mail']}% des mails générés. " "Si ces modifications sont répétitives, signalez-le au référent qualité : " "le prompt ou la base documentaire peut être ajusté." ) else: st.success( f"Taux de modification mail à {stats['taux_modif_mail']}%, " "indicateur de qualité de génération satisfaisant." ) # === ONGLET 4 : HISTORIQUE GLOBAL DES ARCHIVES === def afficher_onglet_historique(): st.subheader("Historique global des appels archivés") st.caption("Toutes les archives (Corpus B). Vue transverse, tous conseillers, tous clients.") archives = lire_archives() if not archives: st.info("Aucun appel archivé pour le moment. Traitez un appel via l'onglet 'Nouvel appel'.") return # Tableau récapitulatif tableau = [] for a in archives: tableau.append({ "Réf. appel": a.get("id_appel", ""), "Date": a.get("date", "")[:10], "Client": a.get("nom_client", ""), "Réf. client": a.get("ref_client", ""), "Conseiller": a.get("nom_conseiller", ""), "Type": a.get("type_demande", ""), "Langue": a.get("langue_mail", "FR"), "FALC": "Oui" if a.get("mode_falc") else "Non", "Statut": a.get("statut", "") }) st.dataframe(tableau, use_container_width=True) # Détail par appel : sélection puis affichage complet st.markdown("### Détails d'un appel archivé") refs = [a.get("id_appel", "") for a in archives] ref_choisie = st.selectbox( "Sélectionner un appel pour voir les détails", options=[""] + refs, format_func=lambda r: r if r else "(aucun)" ) if ref_choisie: archive = next((a for a in archives if a.get("id_appel") == ref_choisie), None) if archive: col1, col2 = st.columns(2) with col1: st.markdown(f"**Client :** {archive.get('nom_client', '')} ({archive.get('ref_client', '')})") st.markdown(f"**Conseiller :** {archive.get('nom_conseiller', '')}") st.markdown(f"**Type :** {archive.get('type_demande', '')}") with col2: st.markdown(f"**Date :** {archive.get('date', '')[:16].replace('T', ' à ')}") st.markdown(f"**Langue mail :** {archive.get('langue_mail', 'FR')}") st.markdown(f"**Mode FALC :** {'Oui' if archive.get('mode_falc') else 'Non'}") with st.expander("Résumé interne validé", expanded=False): st.markdown(archive.get("resume_valide", "(non disponible)")) with st.expander("E-mail envoyé au client", expanded=True): email_envoye = archive.get("email_envoye", "(non disponible)") # Si multilingue, affichage réponse officielle + paragraphe d'aide if "---" in email_envoye: parties = email_envoye.split("---", 1) col_v1, col_v2 = st.columns(2) with col_v1: st.markdown("**Réponse officielle en français**") st.text(parties[0].strip()) with col_v2: st.markdown(f"**Paragraphe d'aide ({archive.get('langue_mail', '?')})**") st.text(parties[1].strip()) else: st.text(email_envoye) # Téléchargement du PDF si disponible chemin_pdf = archive.get("chemin_pdf") if chemin_pdf and os.path.exists(chemin_pdf): with open(chemin_pdf, "rb") as f: st.download_button( "Télécharger le PDF de cet appel", data=f.read(), file_name=f"{ref_choisie}_resume.pdf", mime="application/pdf", key=f"dl_{ref_choisie}" ) st.markdown("---") # Graphiques agrégés st.markdown("### Indicateurs globaux") col1, col2 = st.columns(2) types = {} langues_count = {} for a in archives: t = a.get("type_demande", "Autre") types[t] = types.get(t, 0) + 1 l = a.get("langue_mail", "FR") langues_count[l] = langues_count.get(l, 0) + 1 with col1: st.markdown("**Par type de demande**") st.bar_chart(types) with col2: st.markdown("**Par langue de mail**") st.bar_chart(langues_count) # === MAIN === def main(): st.set_page_config( page_title="GENAURA - Copilote post-appel", page_icon="📮", layout="wide" ) st.title("GENAURA") st.caption("Copilote post-appel La Poste - Hackathon Women in Gen AI 2026") # Sidebar : chat assistant RAG (Corpus A) afficher_chat_assistant() # Onglets principaux onglets = st.tabs([ "Nouvel appel", "Fiche client", "Tableau de bord", "Historique global" ]) with onglets[0]: afficher_onglet_nouvel_appel() with onglets[1]: afficher_onglet_fiche_client() with onglets[2]: afficher_onglet_dashboard() with onglets[3]: afficher_onglet_historique() if __name__ == "__main__": main()