Create crew_utils.py

crew_utils.py

@@ -0,0 +1,187 @@
+# crew_utils.py
+from crewai import Agent, Crew, Process, Task
+from crewai_tools import SerperDevTool
+from langchain_google_genai import ChatGoogleGenerativeAI
+from pydantic import BaseModel, Field
+from typing import List
+from pdf_tool import PDFTool
+
+# --- Définition des outils ---
+search_tool = SerperDevTool()
+pdf_tool = PDFTool()
+
+# --- Définition du LLM (Gemini) ---
+# Remplace par ta clé API
+api_key = "VOTRE_CLE_API_GOOGLE_AI_STUDIO"
+
+# Vérifie si la clé API est définie
+if not api_key:
+    raise ValueError("La clé API Google AI Studio n'est pas définie. Veuillez la configurer comme variable d'environnement GOOGLE_API_KEY.")
+
+llm = ChatGoogleGenerativeAI(
+    model="gemini-pro", google_api_key=api_key, temperature=0.7
+)
+
+# --- Définition des agents ---
+# Chef de Projet
+project_manager = Agent(
+    role="Chef de Projet",
+    goal="Coordonner la création d'un exposé complet et de haute qualité sur le thème donné.",
+    backstory="Expert en gestion de projet et en coordination d'équipes, capable de diriger des projets complexes jusqu'à leur aboutissement.",
+    verbose=True,
+    llm=llm,
+    allow_delegation=True,
+)
+
+# Planificateur
+planner = Agent(
+    role="Planificateur d'Exposé",
+    goal="Créer un plan détaillé et pertinent pour l'exposé.",
+    backstory="Spécialiste de la structuration de contenu, capable de générer des plans clairs et logiques pour des exposés complexes.",
+    verbose=True,
+    llm=llm,
+)
+
+# Rédacteurs (3 instances pour gérer l'introduction, le développement et la conclusion)
+researcher_intro = Agent(
+    role="Rédacteur Spécialisé en Introduction",
+    goal="Rédiger une introduction captivante pour l'exposé.",
+    backstory="Expert en recherche et en rédaction, capable de produire du contenu informatif et bien écrit sur des sujets variés.",
+    verbose=True,
+    llm=llm,
+    tools=[search_tool],
+)
+
+researcher_dev = Agent(
+    role="Rédacteur Spécialisé en Développement",
+    goal="Rédiger les sections de développement de l'exposé.",
+    backstory="Expert en recherche et en rédaction, capable de produire du contenu informatif et bien écrit sur des sujets variés.",
+    verbose=True,
+    llm=llm,
+    tools=[search_tool],
+)
+
+researcher_conclusion = Agent(
+    role="Rédacteur Spécialisé en Conclusion",
+    goal="Rédiger une conclusion percutante pour l'exposé.",
+    backstory="Expert en recherche et en rédaction, capable de produire du contenu informatif et bien écrit sur des sujets variés.",
+    verbose=True,
+    llm=llm,
+    tools=[search_tool],
+)
+
+# Assembleur
+assembler = Agent(
+    role="Assembleur d'Exposé",
+    goal="Compiler toutes les sections rédigées de l'exposé dans un seul document et générer un document PDF final de qualité professionnelle.",
+    backstory="Expert en mise en forme et en compilation de documents, capable de transformer des contenus distincts en un document final harmonieux et esthétique.",
+    verbose=True,
+    llm=llm,
+    tools=[pdf_tool],
+)
+
+# --- Définition des tâches ---
+# Tâche pour le Planificateur
+create_plan_task = Task(
+    description="""
+        Créez un plan détaillé pour un exposé sur le thème suivant : {topic}.
+        Le plan doit inclure une introduction, plusieurs sections principales (au moins 3), et une conclusion.
+        Assurez-vous que le plan est logique et couvre tous les aspects importants du sujet.
+        Divisez les sections principales en sous-sections si nécessaire pour une meilleure organisation.
+    """,
+    expected_output="Un plan d'exposé structuré avec des titres de sections et sous-sections, en format Markdown.",
+    agent=planner,
+)
+
+class SectionContent(BaseModel):
+    title: str = Field(..., description="Titre de la section")
+    content: str = Field(..., description="Contenu de la section")
+
+# Tâches pour les Rédacteurs (exemple pour 3 sections, à adapter)
+async def write_section_task(
+        researcher: Agent, section_title: str, context: List[Any]
+    ) -> Task:
+    return Task(
+        description=f"""
+            Rédigez la section '{section_title}' de l'exposé en vous basant sur le plan fourni.
+            Effectuez des recherches approfondies en utilisant l'outil de recherche pour enrichir le contenu.
+            La section doit être informative, bien écrite et doit respecter le ton académique d'un exposé.
+            Voici le contexte: {context}
+        """,
+        expected_output=f"Texte complet et bien rédigé pour la section '{section_title}' de l'exposé, au format Markdown.",
+        agent=researcher,
+        tools=[search_tool],
+        output_pydantic=SectionContent,  # Utilisez le modèle Pydantic ici
+        context=context,
+    )
+
+# Tâche pour l'Assembleur
+compile_report_task = Task(
+    description="""
+        Compilez toutes les sections rédigées de l'exposé dans un seul document.
+        Organisez les sections selon le plan fourni par le planificateur.
+        Générez un document PDF final prêt pour la présentation.
+        Assurez-vous que le document est bien structuré, facile à lire et qu'il respecte les conventions d'un exposé académique.
+        Voici le contexte: {context}
+    """,
+    expected_output="Un document PDF complet de l'exposé, prêt à être présenté.",
+    agent=assembler,
+    tools=[pdf_tool],
+)
+
+# --- Orchestration des tâches avec un processus hiérarchique ---
+async def process_section_tasks(
+    manager: Agent,
+    plan: Any, # Remplace Any par le type de retour attendu de la tâche create_plan_task
+    researchers: List[Agent],
+):
+    # Assume que plan est une liste de titres de sections
+    sections = plan.split("\n") # Adapter selon le format réel du plan
+    section_tasks: List[Task] = []
+
+    for i, section in enumerate(sections):
+        if section != "":
+            researcher = researchers[i % len(researchers)]
+            section_task = await write_section_task(
+                researcher=researcher,
+                section_title=section.strip(),
+                context=[create_plan_task],
+            )
+            section_tasks.append(section_task)
+
+    return section_tasks
+
+class ExposeCrew(Crew):
+    def __init__(self, agents, tasks, process, manager_llm, verbose):
+        super().__init__(agents=agents, tasks=tasks, process=process, manager_llm=manager_llm, verbose=verbose)
+
+    async def kickoff(self, inputs: dict = {}) -> str:
+        # Exécuter la tâche de création du plan
+        plan = await create_plan_task.execute(context=inputs)
+        print(f"Plan de l'exposé : {plan}")
+
+        # Créer et exécuter les tâches de rédaction de section en parallèle
+        section_tasks = await process_section_tasks(
+            manager=self.manager_llm,
+            plan=plan,
+            researchers=[researcher_intro, researcher_dev, researcher_conclusion],
+        )
+
+        section_outputs = await asyncio.gather(*[task.execute() for task in section_tasks])
+
+        print(f"Sections rédigées : {section_outputs}")
+
+        # Mettre à jour la description de la tâche de compilation avec les sections rédigées
+        compile_report_task.description = f"""
+            Compilez toutes les sections rédigées de l'exposé dans un seul document.
+            Organisez les sections selon le plan fourni par le planificateur :
+            {plan}
+            Sections rédigées :
+            {section_outputs}
+            Générez un document PDF final prêt pour la présentation.
+        """
+        compile_report_task.context = section_tasks
+
+        # Exécuter la tâche de compilation
+        result = await compile_report_task.execute(context=section_outputs)
+        return result