import asyncio from datetime import timedelta from agno.agent import Agent from agno.knowledge.pdf import PDFKnowledgeBase, PDFReader from agno.vectordb.qdrant import Qdrant from agno.memory.v2.memory import Memory from agno.document.chunking.agentic import AgenticChunking from agno.models.openai import OpenAIChat from agno.tools.reasoning import ReasoningTools import gradio as gr import json import pandas as pd from docx import Document import tempfile from weasyprint import HTML import tempfile import markdown as md import os import tempfile from PIL import Image import io from dotenv import load_dotenv load_dotenv() # Initialisation Qdrant vector_db = Qdrant( collection="directives_16J", url=os.getenv("QDRANT_URL"), api_key=os.getenv("QDRANT_API_KEY") ) # Charger tous les PDF du dossier dans la base de connaissance knowledge_base = PDFKnowledgeBase( #path="/home/fallou/PROJECT/Repo-IA/AGNO/directive", reader=PDFReader(chunk=True), vector_db=vector_db, chunking_strategy=AgenticChunking(), ) agent = Agent( model=OpenAIChat(id="gpt-4o"), knowledge=knowledge_base, search_knowledge=True, debug_mode=False, description=( "Agent polyvalent spécialisé dans l'analyse, la génération et la vérification des directives de suivi des projets et programmes de l'État du Sénégal. " "Il combine les rôles suivants : analyse des requêtes, recherche contextuelle, extraction des rôles, rédaction structurée et contrôle de conformité." ), instructions=[ # === CONTEXTE GLOBAL === "### Contexte Global\n" "- Les directives proviennent de toutes les sources officielles : Conseils Présidentiels, Conseils des Ministres, Conseils Interministériels et réunions gouvernementales.\n" "- Extraire **toutes les directives** présentes, même implicites ou partiellement mentionnées.\n" "- L’objectif est de fournir une base exhaustive et directement exploitable pour le suivi des projets et programmes de l’État.\n", # === TÂCHE 1 : RECHERCHE ET EXTRACTION === "### Tâche : Extraction exhaustive des directives\n" "- Identifier toutes les directives liées à `{query}` ou aux projets/programmes mentionnés dans les documents.\n" "- Ne pas omettre les directives secondaires ou implicites (ex : relances, rappels, suivis techniques).\n" "- Lister chaque directive de manière indépendante.\n", # === FORMAT DE SORTIE OBLIGATOIRE === "### Format de sortie obligatoire\n" "- La sortie doit être **en Markdown structuré** avec deux sections principales :\n" "1. **Résumé Global** : courte synthèse expliquant le contexte et nombre de directives trouvées.\n" "2. **Détails des Directives** sous forme de tableau Markdown avec colonnes suivantes :\n" "| Autorité émettrice | Source | Date de publication | Description détaillée de la directive | Action(s) à entreprendre | Structures responsables | Parties prenantes | Statut | Priorité | Date limite | Actions spécifiques du BOCS |\n" "- Chaque ligne représente une directive.\n" "- Les dates doivent être formatées `JJ Mois AAAA` .\n" "- Ne pas inclure de JSON ou code brut, uniquement résumé + tableau Markdown.\n", # === TÂCHE 2 : MÉTADONNÉES ET STRUCTURATION === "### Tâche : Structuration des directives\n" "- Pour chaque directive, fournir :\n" " 1. **Autorité émettrice** : Auteur de la directive (PR, PM, SGG, Ministère, etc.)\n" " 2. **Source** (type de conseil ou réunion)\n" " 3. **Date de publication du document** (ou 'Date du conseil')\n" " 4. **Description détaillée** de la directive\n" " 5. **Actions à entreprendre** (liste détaillée)\n" " 6. **Structures responsables** (entités principales)\n" " 7. **Parties prenantes** (internes/externes)\n" " 8. **Statut** (En cours, Démarré, Non démarré)\n" " 9. **Priorité** (Très Haute, Haute, Moyenne, Basse)\n" " 10. **Date limite** (ou 'Meilleurs delais')\n" " 11. **Actions spécifiques du BOCS** (toujours présentes, détaillées)\n", # === TÂCHE 4 : VÉRIFICATION === "### Vérification\n" "- Vérifier que chaque directive contient **tous les champs**.\n" "- Si une information manque au niveau de la date limite, préciser 'Meilleurs delais'.\n" "- Garantir que la structure est prête à être utilisée sans retravail manuel." ], tools=[ReasoningTools(add_instructions=True)], add_history_to_messages=False, ) # Vérifier si la collection Qdrant contient déjà des points def is_collection_indexed(vector_db): try: info = vector_db.client.count(collection_name=vector_db.collection) return info.count > 0 except Exception: return False # Charger la base de connaissance seulement si elle n'est pas déjà indexée if not is_collection_indexed(vector_db): agent.knowledge.load(recreate=False) # ------------------------------- # Fonction principale : appel agent avec agent.run() # ------------------------------- def generate_directives(query: str): try: prompt = f"Analyse toutes les directives en details en occurence avec la question du user : {query}" result = agent.run(prompt) content = getattr(result, "content", None) if hasattr(content, "dict") and callable(content.dict): data = content.dict() elif isinstance(content, dict): data = content elif isinstance(content, str): return f"Réponse brute :\n\n{content}" else: return f"Type inattendu pour content : {type(content)}\nValeur : {content}" resume = data.get("resume") directives = data.get("directives") if not resume or not directives: return "Le JSON ne contient pas 'resume' ou 'directives'." # Générer seulement du Markdown markdown_output = f"## Résumé\n- **Nombre de directives** : {resume.get('nombre_directives', 'Inconnu')}\n\n" markdown_output += f"> {resume.get('contexte', '')}\n\n" markdown_output += "## Directives détaillées\n" for idx, directive in enumerate(directives, start=1): markdown_output += f"### Directive {idx}\n" for key, value in directive.items(): if isinstance(value, list): value = ", ".join(value) markdown_output += f"- **{key}** : {value}\n" markdown_output += "\n" return markdown_output except Exception as e: return f"Erreur inattendue : {str(e)}" # --- Convertir Markdown en PDF --- def export_to_pdf(markdown_text): if not markdown_text: return None # Convertir Markdown en HTML html_content = md.markdown(markdown_text) tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") HTML(string=html_content).write_pdf(tmp.name) return tmp.name def export_to_html(markdown_text): if not markdown_text: return None html_content = md.markdown(markdown_text) tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".html") with open(tmp.name, "w", encoding="utf-8") as f: f.write(f"{html_content}") return tmp.name # ------------------------------- # Interface Gradio avec deux onglets # ------------------------------- with gr.Blocks() as demo: gr.Markdown("# 🏛️ Assistant Suivi des Directives") # Onglet Chat with gr.Tab("Chat"): with gr.Row(): query_input = gr.Textbox( label="Votre requête", placeholder="Ex: Détails spécifiques sur des directives individuelles. ", lines=3 ) submit_btn = gr.Button("Générer Directives") output_md = gr.Markdown(label="Résultat") # Ici on relie directement la fonction submit_btn.click( fn=generate_directives, inputs=query_input, outputs=output_md ) # Onglet Export (toujours à l’intérieur du même Blocks) with gr.Tab("Export"): gr.Markdown("### Exportez les directives générées") export_pdf_btn = gr.Button("Télécharger PDF (.pdf)") export_html_btn = gr.Button("Télécharger HTML (.html)") pdf_file = gr.File(label="Fichier PDF", interactive=False) html_file = gr.File(label="Fichier HTML", interactive=False) export_pdf_btn.click(fn=export_to_pdf, inputs=output_md, outputs=pdf_file) export_html_btn.click(fn=export_to_html, inputs=output_md, outputs=html_file) demo.launch(server_name="0.0.0.0", server_port=7860)