Create cv_agents.py

src/agents/cv_agents.py

@@ -0,0 +1,192 @@
+import json
+import logging
+from typing import Dict, Any, List
+from crewai import Agent, Task, Crew, Process
+
+logger = logging.getLogger(__name__)
+
+class CVAgentOrchestrator:
+    def __init__(self, llm):
+        self.llm = llm
+        self._create_agents()
+    
+    def _create_agents(self):
+        self.section_splitter = Agent(
+            role="Analyseur de Structure de CV",
+            goal="Découper intelligemment un CV en sections thématiques",
+            backstory="Expert en analyse documentaire spécialisé dans la reconnaissance de structures de CV.",
+            verbose=False,
+            llm=self.llm
+        )
+        
+        self.contact_extractor = Agent(
+            role="Extracteur d'informations de contact",
+            goal="Extraire les coordonnées du candidat",
+            backstory="Expert en extraction d'informations de contact avec précision.",
+            verbose=False,
+            llm=self.llm
+        )
+        
+        self.skills_extractor = Agent(
+            role="Extracteur de compétences",
+            goal="Identifier hard skills et soft skills",
+            backstory="Spécialiste en identification de compétences techniques et comportementales.",
+            verbose=False,
+            llm=self.llm
+        )
+        
+        self.experience_extractor = Agent(
+            role="Extracteur d'expériences",
+            goal="Extraire les expériences professionnelles",
+            backstory="Expert en analyse de parcours professionnels.",
+            verbose=False,
+            llm=self.llm
+        )
+        
+        self.project_extractor = Agent(
+            role="Extracteur de projets",
+            goal="Identifier projets professionnels et personnels",
+            backstory="Spécialiste en identification de projets significatifs.",
+            verbose=False,
+            llm=self.llm
+        )
+        
+        self.education_extractor = Agent(
+            role="Extracteur de formations",
+            goal="Extraire formations et diplômes",
+            backstory="Expert en analyse de parcours académiques.",
+            verbose=False,
+            llm=self.llm
+        )
+        
+        self.reconversion_detector = Agent(
+            role="Détecteur de reconversion",
+            goal="Analyser les changements de carrière",
+            backstory="Conseiller d'orientation expert en transitions de carrière.",
+            verbose=False,
+            llm=self.llm
+        )
+        
+        self.profile_builder = Agent(
+            role="Constructeur de profil",
+            goal="Assembler le profil candidat final",
+            backstory="Expert en structuration de données JSON.",
+            verbose=False,
+            llm=self.llm
+        )
+    
+    def split_cv_sections(self, cv_content: str) -> Dict[str, str]:
+        task = Task(
+            description=f"Analyser ce CV et l'organiser en sections: {cv_content}",
+            expected_output="""JSON avec sections: contact, experiences, projects, education, skills, other""",
+            agent=self.section_splitter
+        )
+        
+        crew = Crew(
+            agents=[self.section_splitter],
+            tasks=[task],
+            process=Process.sequential,
+            verbose=False,
+            telemetry=False
+        )
+        
+        result = crew.kickoff()
+        return self._parse_sections_result(result)
+    
+    def extract_all_sections(self, sections: Dict[str, str]) -> Dict[str, Any]:
+        tasks = self._create_extraction_tasks(sections)
+        
+        crew = Crew(
+            agents=[
+                self.contact_extractor,
+                self.skills_extractor,
+                self.experience_extractor,
+                self.project_extractor,
+                self.education_extractor,
+                self.reconversion_detector,
+                self.profile_builder
+            ],
+            tasks=tasks,
+            process=Process.sequential,
+            verbose=False,
+            telemetry=False
+        )
+        
+        result = crew.kickoff(inputs=sections)
+        return self._parse_final_result(result)
+    
+    def _create_extraction_tasks(self, sections: Dict[str, str]) -> List[Task]:
+        contact_task = Task(
+            description="Extraire nom, email, téléphone, localisation de: {contact}",
+            expected_output='{"nom": "...", "email": "...", "numero_de_telephone": "...", "localisation": "..."}',
+            agent=self.contact_extractor
+        )
+        
+        skills_task = Task(
+            description="Extraire compétences de: {experiences} {projects} {skills}",
+            expected_output='{"hard_skills": [...], "soft_skills": [...]}',
+            agent=self.skills_extractor,
+            context=[contact_task]
+        )
+        
+        experience_task = Task(
+            description="Extraire expériences de: {experiences}",
+            expected_output='[{"Poste": "...", "Entreprise": "...", "start_date": "...", "end_date": "...", "responsabilités": [...]}]',
+            agent=self.experience_extractor,
+            context=[skills_task]
+        )
+        
+        project_task = Task(
+            description="Extraire projets de: {projects} {experiences}",
+            expected_output='{"professional": [...], "personal": [...]}',
+            agent=self.project_extractor,
+            context=[experience_task]
+        )
+        
+        education_task = Task(
+            description="Extraire formations de: {education}",
+            expected_output='[{"degree": "...", "institution": "...", "start_date": "...", "end_date": "..."}]',
+            agent=self.education_extractor,
+            context=[project_task]
+        )
+        
+        reconversion_task = Task(
+            description="Analyser reconversion basée sur les expériences extraites",
+            expected_output='{"reconversion_analysis": {"is_reconversion": true/false, "analysis": "..."}}',
+            agent=self.reconversion_detector,
+            context=[education_task]
+        )
+        
+        profile_task = Task(
+            description="Assembler profil final JSON avec clé 'candidat'",
+            expected_output='{"candidat": {...}}',
+            agent=self.profile_builder,
+            context=[reconversion_task]
+        )
+        
+        return [contact_task, skills_task, experience_task, project_task, education_task, reconversion_task, profile_task]
+    
+    def _parse_sections_result(self, result) -> Dict[str, str]:
+        result_str = result.raw if hasattr(result, 'raw') else str(result)
+        
+        if '```json' in result_str:
+            result_str = result_str.split('```json')[1].split('```')[0].strip()
+        elif '```' in result_str:
+            parts = result_str.split('```')
+            if len(parts) >= 3:
+                result_str = parts[1].strip()
+        
+        parsed = json.loads(result_str)
+        return parsed
+    
+    def _parse_final_result(self, result) -> Dict[str, Any]:
+        result_str = result.raw if hasattr(result, 'raw') else str(result)
+        
+        if '```json' in result_str:
+            result_str = result_str.split('```json')[1].split('```')[0].strip()
+        elif '```' in result_str:
+            parts = result_str.split('```')
+            if len(parts) >= 3:
+                result_str = parts[1].strip()
+        
+        return json.loads(result_str)