Simplification code IA
Creado el: 16 de enero de 2025
Creado el: 16 de enero de 2025
peux tu me refaire ce code ? je veux qu'il sois plus simple a comprendre pour que je puisse le modifie plus facilement et avec des commentaire en français
from typing import Dict, Any, List, Optional, Union
from pathlib import Path
from rich.console import Console
from rich.panel import Panel
from rich.markdown import Markdown
from rich.prompt import Confirm, Prompt
from rich.progress import Progress
from rich.table import Table
import asyncio
import json
import re
console = Console()
class AIEngineer:
"""An autonomous AI Software Engineer that can plan and execute complex development tasks."""
textdef __init__(self, groq_service, file_manager, context_manager, settings_manager): self.groq = groq_service self.file_manager = file_manager self.context_manager = context_manager self.settings = settings_manager # Thinking patterns for different tasks self.thought_patterns = { "analyze": "Let me analyze this request...", "plan": "I'll create a plan to implement this...", "code": "I'll write the code with best practices...", "refactor": "This code could be improved by...", "explain": "Let me explain what I'm doing...", "execute" : "execute the code" } async def process_request(self, request: str) -> Dict[str, Any]: """Process a user request with detailed thinking and explanation.""" try: # 1. Initial Analysis console.print("\n[bold blue]🤔 Analyzing request...[/bold blue]") # 2. Get and show approach approach = await self._determine_approach(request) # Show analysis console.print(Panel( f"""[bold]Request:[/bold] {approach['analysis']['request']} [bold]Description:[/bold] {approach['analysis']['description']} [bold]Complexity:[/bold] {approach['analysis']['complexity']} [bold]Estimated time:[/bold] {approach['analysis']['estimated_time']}""", title="Analysis" )) # Show steps console.print("\n[bold green]📋 Implementation Plan:[/bold green]") for i, step in enumerate(approach["steps"], 1): console.print(f"{i}. {step}") # Show challenges if any if approach.get("challenges"): # moi console.print("voulez vous lancez un ultra mega think ?") console.print("\n[bold yellow]⚠️ Potential Challenges:[/bold yellow]") for challenge in approach["challenges"]: console.print(f"• {challenge}") if not Confirm.ask("\nShould I proceed with this plan?"): return {"success": False, "message": "Plan rejected by user"} # 3. Execute with Explanations result = await self._execute_with_explanation(approach, request) # 4. Summarize Results self._display_summary(result) return result except Exception as e: console.print(f"[red]Error: {str(e)}[/red]") return {"success": False, "message": str(e)} async def _determine_approach(self, request: str) -> Dict[str, Any]: """Think through and determine the best approach for the request.""" try: # Get current context context = await self.groq.get_project_context() # Ask AI to analyze and create approach with strict JSON format prompt = f"""Analyze this request and create a detailed implementation plan. IMPORTANT: Your response must be ONLY valid JSON in this exact format: {{ "analysis": {{ "request": "{request}", "description": "What needs to be done", "complexity": "simple/medium/complex", "estimated_time": "X hours/minutes" }}, "steps": [ "Step 1: Description", "Step 2: Description" ], "challenges": [ "Challenge 1", "Challenge 2" ], "changes": [ {{ "type": "create", "path": "path/to/file.py", "description": "What to create", "step": 1 }}, {{ "type": "modify", "path": "existing/file.py", "description": "What to change", "step": 2 }} ], }} Request: {request} Current Project Context: {json.dumps(context, indent=2)}""" # Add system prompt to enforce JSON response system_prompt = """You are a senior software architect. CRITICAL: Return ONLY valid JSON matching the exact format shown. Do not include any other text, markdown, or explanations.""" response = await self.groq.send_message( prompt, system_prompt=system_prompt ) # Clean and parse response try: # Remove any non-JSON content response = response.strip() # Remove markdown code blocks if present response = re.sub(r'^```json\s*|\s*```$', '', response, flags=re.MULTILINE) # Find the JSON object json_match = re.search(r'\{[\s\S]*\}', response) if not json_match: raise ValueError("No JSON object found in response") plan = json.loads(json_match.group(0)) # Validate required fields required_fields = ['analysis', 'steps', 'changes'] missing_fields = [f for f in required_fields if f not in plan] if missing_fields: raise ValueError(f"Missing required fields: {', '.join(missing_fields)}") return plan except (json.JSONDecodeError, ValueError) as e: # Log the error and response for debugging console.print("[red]Error parsing AI response[/red]") console.print("\n[yellow]Raw response:[/yellow]") console.print(response) console.print(f"\n[red]Error: {str(e)}[/red]") # Create a fallback plan return { "analysis": { "request": request, "description": "Fallback plan due to parsing error", "complexity": "unknown", "estimated_time": "unknown" }, "steps": ["Analyze request", "Implement solution"], "challenges": ["Error parsing AI response"], "changes": [ { "type": "analyze", "path": "unknown", "description": "Original request: " + request, "step": 1 } ], "testing": ["Verify implementation"] } except Exception as e: console.print(f"[red]Error in determine_approach: {str(e)}[/red]") raise async def _execute_with_explanation(self, approach: Dict[str, Any], request: str) -> Dict[str, Any]: """Execute the approach while explaining each step.""" results = [] with Progress() as progress: task = progress.add_task("Implementing...", total=len(approach["steps"])) for step_num, step in enumerate(approach["steps"], 1): # Explain what we're doing console.print(f"\n[bold blue]Step {step_num}:[/bold blue] {step}") # Execute the step if step_num < len(approach["steps"]): if not Confirm.ask("Should I proceed with this step?"): return { "success": False, "message": f"Execution stopped at step {step_num}" } # Execute changes for this step changes = [c for c in approach["changes"] if c.get("step", step_num) == step_num] for change in changes: result = await self._execute_change(change, request) results.append(result) # Explain what we did if result["success"]: console.print(f"[green]✓[/green] {result['message']}") else: console.print(f"[red]✗[/red] {result['message']}") progress.advance(task) return { "success": all(r["success"] for r in results), "results": results, "approach": approach } async def _execute_change(self, change: Dict[str, Any], context: str) -> Dict[str, Any]: """Execute a single change with explanation.""" try: change_type = change["type"] file_path = change["path"] if change_type == "create": # Generate and create new file content = await self._generate_file_content(file_path, change, context) return await self.file_manager.create_file(file_path, content) elif change_type == "modify": # Modify existing file current_content = self.file_manager.read_file(file_path) new_content = await self._generate_file_content( file_path, change, context, current_content) return await self.file_manager.save_file(file_path, new_content) elif change_type == "delete": return await self.file_manager.delete_file(file_path) except Exception as e: return { "success": False, "message": f"Error in {change_type} for {file_path}: {str(e)}" } def _display_summary(self, result: Dict[str, Any]) -> None: """Display a summary of all changes made.""" if not result["success"]: console.print("[red]❌ Implementation failed[/red]") return console.print("[green]✅ Implementation completed successfully[/green]") # Show changes made table = Table(title="Changes Made") table.add_column("File", style="cyan") table.add_column("Action", style="green") table.add_column("Status", style="yellow") for r in result["results"]: table.add_row( r.get("file", "unknown"), r.get("action", "unknown"), "✓" if r["success"] else "✗" ) console.print(table)
Voici une version plus simple et plus concise du code, accompagnée de commentaires en français pour faciliter sa compréhension et sa modification.
Vous pouvez l’adapter à votre convenance en fonction de vos besoins.
pythonfrom typing import Dict, Any, List, Optional, Union from pathlib import Path import asyncio import json import re # Bibliothèques Rich pour un affichage amélioré dans le terminal from rich.console import Console from rich.panel import Panel from rich.prompt import Confirm from rich.progress import Progress from rich.table import Table console = Console() class AIEngineer: """ IA Ingénieur Logiciel autonome capable de planifier et d'exécuter des tâches de développement. Cette version du code est simplifiée pour être plus lisible et modifiable. """ def __init__(self, groq_service, file_manager, context_manager, settings_manager): """ Initialisation de l'IAEngineer avec différents services nécessaires. :param groq_service: Service d'interaction AI (ex.: GPT) :param file_manager: Gestionnaire de fichiers (création, lecture, écriture...) :param context_manager: Gestionnaire du contexte de projet :param settings_manager: Gestionnaire des paramètres """ self.groq = groq_service self.file_manager = file_manager self.context_manager = context_manager self.settings = settings_manager async def process_request(self, request: str) -> Dict[str, Any]: """ Méthode principale pour traiter la requête utilisateur. 1. Analyse la requête et génère un plan. 2. Affiche le plan et confirme avec l'utilisateur. 3. Exécute le plan (création ou modification de fichiers). 4. Affiche un résumé des actions menées. :param request: La demande de l'utilisateur sous forme de chaîne de caractères. :return: Un dictionnaire contenant le succès ou l'échec, et les détails des actions. """ try: # 1. Déterminer l'approche à adopter pour la requête console.print("\n[bold blue]Étape 1 : Analyse de la requête...[/bold blue]") approche = await self._determine_approach(request) # Affiche un résumé de l'analyse self._afficher_analyse(approche) # 2. Confirmation de l'utilisateur pour continuer if not Confirm.ask("\nSouhaitez-vous procéder à l'exécution du plan ?"): return {"success": False, "message": "Plan annulé par l'utilisateur"} # 3. Exécuter le plan console.print("\n[bold blue]Étape 2 : Exécution du plan...[/bold blue]") resultats = await self._executer_plan(approche, request) # 4. Afficher le résumé des opérations self._afficher_resume(resultats) return resultats except Exception as e: console.print(f"[red]Erreur lors du traitement: {str(e)}[/red]") return {"success": False, "message": str(e)} async def _determine_approach(self, request: str) -> Dict[str, Any]: """ Analyse la requête et détermine l'approche (plan) à adopter. Cette méthode interroge le service AI pour obtenir un plan structuré (format JSON). :param request: La demande de l'utilisateur. :return: Un dictionnaire représentant l'approche (plan, analyses, étapes...). """ try: # Récupère le contexte actuel du projet contexte = await self.groq.get_project_context() # Prépare la demande à envoyer à l'IA prompt = f"""Analyse cette requête et produis un plan détaillé au format JSON (sans texte supplémentaire). Format attendu : {{ "analysis": {{ "request": "{request}", "description": "Ce qu'il faut faire", "complexity": "simple/medium/complex", "estimated_time": "Durée estimée" }}, "steps": [ "Étape 1", "Étape 2" ], "challenges": [ "Challenge 1", "Challenge 2" ], "changes": [ {{ "type": "create", "path": "nouveau/fichier.py", "description": "Rôle du nouveau fichier", "step": 1 }}, {{ "type": "modify", "path": "existant/fichier.py", "description": "Modifications à effectuer", "step": 2 }} ] }} Requête: {request} Contexte du projet: {json.dumps(contexte, indent=2)} """ # Contrainte pour forcer le retour d'un JSON strict system_prompt = ( "Tu es un architecte logiciel expérimenté. " "Retourne UNIQUEMENT du JSON valide correspondant au format spécifié." ) # Envoi de la requête à l'IA reponse = await self.groq.send_message(prompt, system_prompt=system_prompt) # Nettoyage de la réponse pour extraire le JSON plan = self._extraire_json_depuis_chaine(reponse) return plan except Exception as e: console.print(f"[red]Erreur dans _determine_approach: {str(e)}[/red]") # Si le parsing échoue, on renvoie un plan de secours return { "analysis": { "request": request, "description": "Plan de secours suite à une erreur de parsing", "complexity": "inconnu", "estimated_time": "inconnu" }, "steps": ["Analyse", "Mise en œuvre"], "challenges": [f"Erreur : {str(e)}"], "changes": [] } def _extraire_json_depuis_chaine(self, chaine: str) -> Dict[str, Any]: """ Extrait un objet JSON depuis une chaîne en supprimant tout texte superflu. :param chaine: Réponse brute contenant potentiellement le JSON. :return: Dictionnaire Python issu du JSON. """ try: # Supprime d'éventuels "```json" ou "```" propre = re.sub(r'^```json\s*|\s*```$', '', chaine.strip(), flags=re.MULTILINE) # Recherche l'objet JSON avec une expression régulière correspondance = re.search(r'\{[\s\S]*\}', propre) if not correspondance: raise ValueError("Aucun objet JSON trouvé") return json.loads(correspondance.group(0)) except Exception as e: console.print("[red]Impossible d'extraire le JSON valide[/red]") raise ValueError(f"Erreur de parsing : {str(e)}") async def _executer_plan(self, approche: Dict[str, Any], request: str) -> Dict[str, Any]: """ Exécute le plan en parcourant les différentes étapes et les modifications à effectuer. :param approche: Le plan généré par l'IA (analyse, steps, changes, etc.). :param request: La requête d'origine de l'utilisateur. :return: Un dictionnaire avec le succès global et le détail des actions. """ resultats = [] etapes = approche.get("steps", []) modifications = approche.get("changes", []) # Utilisation d'une barre de progression pour suivre les étapes with Progress() as progress: tache = progress.add_task("Exécution du plan...", total=len(etapes)) for index_etape, etape in enumerate(etapes, start=1): console.print(f"\n[bold green]Étape {index_etape}:[/bold green] {etape}") # Demande à l'utilisateur s'il veut procéder à chaque étape if not Confirm.ask("Continuer cette étape ?"): return { "success": False, "message": f"Exécution interrompue à l'étape {index_etape}", "results": resultats } # Exécute les changements associés à l'étape courante changements_etape = [ c for c in modifications if c.get("step") == index_etape ] for changement in changements_etape: resultat = await self._executer_changement(changement, request) resultats.append(resultat) if resultat["success"]: console.print(f"[green]✓[/green] {resultat['message']}") else: console.print(f"[red]✗[/red] {resultat['message']}") progress.advance(tache) # Vérifie si tous les changements ont été exécutés avec succès success_global = all(r["success"] for r in resultats) return { "success": success_global, "results": resultats, "approach": approche } async def _executer_changement(self, changement: Dict[str, Any], contexte: str) -> Dict[str, Any]: """ Exécute un seul changement (create, modify, delete...) sur un fichier. :param changement: Détails du changement (type, path, description, step). :param contexte: Contexte ou requête d'origine (optionnel, pour la génération du contenu). :return: Un dictionnaire avec le succès et un message. """ type_changement = changement.get("type") chemin_fichier = changement.get("path") try: if type_changement == "create": # Génère le contenu et crée un nouveau fichier contenu = await self._generer_contenu(chemin_fichier, changement, contexte) return await self.file_manager.create_file(chemin_fichier, contenu) elif type_changement == "modify": # Lit l'existant et génère le nouveau contenu contenu_actuel = self.file_manager.read_file(chemin_fichier) nouveau_contenu = await self._generer_contenu( chemin_fichier, changement, contexte, contenu_actuel ) return await self.file_manager.save_file(chemin_fichier, nouveau_contenu) elif type_changement == "delete": return await self.file_manager.delete_file(chemin_fichier) else: return { "success": False, "message": f"Type de changement inconnu: {type_changement}", "file": chemin_fichier, "action": type_changement } except Exception as e: return { "success": False, "message": f"Erreur lors de l'opération {type_changement} sur {chemin_fichier}: {str(e)}", "file": chemin_fichier, "action": type_changement } async def _generer_contenu( self, chemin_fichier: str, changement: Dict[str, Any], contexte: str, contenu_actuel: Optional[str] = None ) -> str: """ Méthode d'exemple pour générer ou modifier le contenu d'un fichier en fonction de la requête et d'un contenu existant. :param chemin_fichier: Chemin du fichier à créer ou modifier. :param changement: Détails du changement (description...). :param contexte: Contexte ou requête d'origine. :param contenu_actuel: Contenu actuel du fichier (en cas de modification). :return: Le nouveau contenu généré. """ # Pour simplifier, on renvoie juste un texte générique. # Vous pouvez y inclure un appel à self.groq.send_message(...) pour générer un contenu élaboré. nouvelle_description = changement.get("description", "Pas de description") if contenu_actuel: return contenu_actuel + f"\n# Modification: {nouvelle_description}\n" else: return f"# Nouveau fichier\n# Description: {nouvelle_description}\n" def _afficher_analyse(self, approche: Dict[str, Any]) -> None: """ Affiche l'analyse et le plan générés par l'IA. :param approche: Dictionnaire contenant l'analyse, les étapes, etc. """ analysis = approche.get("analysis", {}) console.print(Panel( f"[bold]Requête :[/bold] {analysis.get('request', '')}\n" f"[bold]Description :[/bold] {analysis.get('description', '')}\n" f"[bold]Complexité :[/bold] {analysis.get('complexity', '')}\n" f"[bold]Temps estimé :[/bold] {analysis.get('estimated_time', '')}", title="Analyse" )) console.print("\n[bold green]Plan d'implémentation :[/bold green]") for i, step in enumerate(approche.get("steps", []), 1): console.print(f"{i}. {step}") if approche.get("challenges"): console.print("\n[bold yellow]Points de vigilance :[/bold yellow]") for challenge in approche["challenges"]: console.print(f"• {challenge}") def _afficher_resume(self, resultat: Dict[str, Any]) -> None: """ Affiche un tableau récapitulatif des fichiers modifiés ou créés. :param resultat: Le dictionnaire résultant de l'exécution du plan. """ if not resultat["success"]: console.print("[red]❌ Échec de l'exécution[/red]") else: console.print("[green]✅ Exécution terminée avec succès[/green]") # Création d'un tableau pour présenter les changements effectués tableau = Table(title="Bilan des changements") tableau.add_column("Fichier", style="cyan") tableau.add_column("Action", style="green") tableau.add_column("Statut", style="yellow") for r in resultat.get("results", []): fichier = r.get("file", "inconnu") action = r.get("action", "inconnu") statut = "✓" if r["success"] else "✗" tableau.add_row(fichier, action, statut) console.print(tableau)
Vous disposez ainsi d’une base plus facile à maintenir et à faire évoluer.
Il vous reste à personnaliser chaque section pour qu’elle réponde exactement à vos besoins (génération de contenu, structure de projet, validations, etc.).