AutoGPT et BabyAGI : Agents Autonomes

User: Recherche les meilleurs restaurants à Paris
Agent: [Recherche] → [Retourne liste]
✅ Terminé en 30 secondes

User: Organise un voyage à Paris pour 3 jours
Agent:
  1. Recherche vols Paris
  2. Compare prix hôtels
  3. Trouve restaurants
  4. Crée itinéraire jour par jour
  5. Réserve (si autorisé)
  6. Génère document récapitulatif
✅ Terminé en 2 heures, 150+ appels LLM

from typing import List, Dict
import openai
import json

class AutoGPT:
    """Implémentation simplifiée d'AutoGPT"""

    def __init__(
        self,
        goal: str,
        max_iterations: int = 50,
        budget_dollars: float = 10.0
    ):
        self.goal = goal
        self.max_iterations = max_iterations
        self.budget_dollars = budget_dollars
        self.spent = 0.0

        self.memory = []  # Mémoire court-terme
        self.task_list = []  # Liste de tâches
        self.completed_tasks = []  # Tâches complétées

    def run(self) -> str:
        """Boucle principale autonome"""

        print(f"🎯 Objectif : {self.goal}")
        print(f"💰 Budget : ${self.budget_dollars}")

        # Étape 1 : Créer le plan initial
        self._create_initial_plan()

        iteration = 0

        while iteration < self.max_iterations and self.spent < self.budget_dollars:
            iteration += 1
            print(f"\n{'='*60}")
            print(f"🔄 Itération {iteration}/{self.max_iterations}")
            print(f"💸 Dépensé : ${self.spent:.2f} / ${self.budget_dollars}")
            print(f"{'='*60}")

            # Étape 2 : Prendre la prochaine tâche
            if not self.task_list:
                print("✅ Toutes les tâches sont complétées !")
                break

            current_task = self.task_list.pop(0)
            print(f"\n📋 Tâche actuelle : {current_task}")

            # Étape 3 : Exécuter la tâche
            result = self._execute_task(current_task)
            print(f"✅ Résultat : {result[:150]}...")

            # Étape 4 : Auto-évaluation
            evaluation = self._evaluate_progress()
            print(f"🔍 Évaluation : {evaluation['status']}")

            # Étape 5 : Ajuster le plan si nécessaire
            if evaluation["should_add_tasks"]:
                new_tasks = evaluation.get("new_tasks", [])
                self.task_list.extend(new_tasks)
                print(f"➕ {len(new_tasks)} nouvelles tâches ajoutées")

            # Étape 6 : Vérifier si l'objectif est atteint
            if evaluation.get("goal_achieved"):
                print(f"\n🎉 Objectif atteint !")
                break

        # Générer le rapport final
        return self._generate_final_report()

    def _create_initial_plan(self):
        """Crée le plan de tâches initial"""
        print("\n📝 Création du plan initial...")

        prompt = f"""Tu es AutoGPT, un agent autonome.

Objectif final : {self.goal}

Crée une liste de 5-10 tâches concrètes et séquentielles pour atteindre cet objectif.

Réponds en JSON : {{"tasks": ["tâche 1", "tâche 2", ...]}}"""

        response = self._call_llm(prompt)

        try:
            plan = json.loads(response)
            self.task_list = plan.get("tasks", [])
            print(f"✅ Plan créé : {len(self.task_list)} tâches")
            for i, task in enumerate(self.task_list, 1):
                print(f"  {i}. {task}")
        except json.JSONDecodeError:
            print("❌ Erreur de parsing du plan")
            self.task_list = [self.goal]  # Fallback

    def _execute_task(self, task: str) -> str:
        """Exécute une tâche"""

        # Contexte depuis la mémoire
        context = "\n".join([
            f"- {t['task']}: {t['result'][:100]}"
            for t in self.completed_tasks[-3:]  # 3 dernières tâches
        ])

        prompt = f"""Tu es AutoGPT. Exécute cette tâche.

Objectif global : {self.goal}

Tâches précédentes :
{context if context else "Aucune"}

Tâche actuelle : {task}

Fournis un résultat concret et actionnable."""

        result = self._call_llm(prompt)

        # Enregistrer
        self.completed_tasks.append({
            "task": task,
            "result": result
        })

        return result

    def _evaluate_progress(self) -> Dict:
        """Évalue la progression vers l'objectif"""

        completed_summary = "\n".join([
            f"- {t['task']}: {t['result'][:80]}"
            for t in self.completed_tasks
        ])

        remaining_summary = "\n".join([
            f"- {task}" for task in self.task_list[:3]
        ])

        prompt = f"""Tu es AutoGPT. Évalue ta progression.

Objectif : {self.goal}

Tâches complétées :
{completed_summary}

Tâches restantes :
{remaining_summary if remaining_summary else "Aucune"}

Évalue :
1. L'objectif est-il atteint ? (oui/non)
2. Faut-il ajouter des tâches ? (oui/non)
3. Si oui, liste les nouvelles tâches

Réponds en JSON :
{{
    "goal_achieved": true/false,
    "should_add_tasks": true/false,
    "new_tasks": ["tâche 1", ...],
    "status": "description de l'état actuel"
}}"""

        response = self._call_llm(prompt)

        try:
            return json.loads(response)
        except json.JSONDecodeError:
            return {
                "goal_achieved": False,
                "should_add_tasks": False,
                "status": "Évaluation en cours"
            }

    def _generate_final_report(self) -> str:
        """Génère un rapport final"""

        summary = "\n\n".join([
            f"**Tâche {i+1}** : {t['task']}\n{t['result']}"
            for i, t in enumerate(self.completed_tasks)
        ])

        prompt = f"""Génère un rapport final résumant le travail accompli.

Objectif initial : {self.goal}

Tâches accomplies :
{summary}

Crée un rapport structuré et professionnel."""

        return self._call_llm(prompt)

    def _call_llm(self, prompt: str) -> str:
        """Appelle le LLM et track les coûts"""

        response = openai.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7
        )

        # Estimer le coût (approximation)
        tokens_used = response.usage.total_tokens
        cost = (tokens_used / 1000) * 0.03  # ~$0.03/1k tokens pour GPT-4

        self.spent += cost

        return response.choices[0].message.content


# ========== EXEMPLE D'UTILISATION ==========

if __name__ == "__main__":
    agent = AutoGPT(
        goal="Recherche et résume les 3 principales tendances en IA pour 2024",
        max_iterations=20,
        budget_dollars=5.0
    )

    final_report = agent.run()

    print(f"\n\n{'='*60}")
    print("📄 RAPPORT FINAL")
    print(f"{'='*60}\n")
    print(final_report)
    print(f"\n💰 Coût total : ${agent.spent:.2f}")

🎯 Objectif : Recherche et résume les 3 principales tendances en IA pour 2024
💰 Budget : $5.0

📝 Création du plan initial...
✅ Plan créé : 6 tâches
  1. Rechercher articles récents sur les tendances IA 2024
  2. Identifier les tendances les plus mentionnées
  3. Approfondir la recherche sur chaque tendance
  4. Analyser l'impact de chaque tendance
  5. Comparer et classer les tendances par importance
  6. Rédiger un résumé structuré

============================================================
🔄 Itération 1/20
💸 Dépensé : $0.00 / $5.0
============================================================

📋 Tâche actuelle : Rechercher articles récents sur les tendances IA 2024
✅ Résultat : J'ai identifié plusieurs sources mentionnant : IA générative, agents autonomes, multimodalité...

🔍 Évaluation : Progression normale, continuer
➕ 0 nouvelles tâches ajoutées

[... itérations 2-6 ...]

🎉 Objectif atteint !

============================================================
📄 RAPPORT FINAL
============================================================

# Les 3 Principales Tendances IA 2024

## IA Générative Multi-Modale {#ia-générative-multi-modale}
[Description détaillée]

## Agents Autonomes {#agents-autonomes}
[Description détaillée]

## IA Edge Computing {#ia-edge-computing}
[Description détaillée]

💰 Coût total : $3.45

from collections import deque
import openai

class BabyAGI:
    """Implémentation simplifiée de BabyAGI"""

    def __init__(self, objective: str, initial_task: str = None):
        self.objective = objective
        self.task_queue = deque()
        self.completed_tasks = []

        # Ajouter la tâche initiale
        if initial_task:
            self.task_queue.append({
                "task_id": 1,
                "task_name": initial_task
            })
        else:
            self.task_queue.append({
                "task_id": 1,
                "task_name": "Créer un plan pour atteindre l'objectif"
            })

        self.next_task_id = 2

    def run(self, max_iterations: int = 10):
        """Boucle principale"""

        iteration = 0

        while iteration < max_iterations and self.task_queue:
            iteration += 1
            print(f"\n{'='*60}")
            print(f"🔄 Itération {iteration}")
            print(f"📋 Tâches en attente : {len(self.task_queue)}")
            print(f"{'='*60}")

            # 1. Prendre la prochaine tâche
            current_task = self.task_queue.popleft()

            print(f"\n▶️ Exécution : {current_task['task_name']}")

            # 2. Exécuter la tâche
            result = self._execute_task(current_task)
            print(f"✅ Résultat : {result[:150]}...")

            # 3. Enregistrer
            self.completed_tasks.append({
                **current_task,
                "result": result
            })

            # 4. Créer de nouvelles tâches basées sur le résultat
            new_tasks = self._create_new_tasks(result, current_task)

            if new_tasks:
                print(f"\n➕ {len(new_tasks)} nouvelles tâches créées")
                for task in new_tasks:
                    print(f"   • {task['task_name']}")

            # 5. Prioriser les tâches
            self._prioritize_tasks()

        print(f"\n\n🏁 Terminé après {iteration} itérations")
        self._print_summary()

    def _execute_task(self, task: Dict) -> str:
        """Exécute une tâche"""

        # Contexte des tâches précédentes
        context = "\n".join([
            f"{t['task_id']}. {t['task_name']}: {t['result'][:80]}"
            for t in self.completed_tasks[-3:]
        ])

        prompt = f"""Tu es un agent d'exécution de tâches.

Objectif global : {self.objective}

Tâches précédentes :
{context if context else "Aucune"}

Tâche actuelle : {task['task_name']}

Exécute cette tâche et fournis un résultat concret."""

        response = openai.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}]
        )

        return response.choices[0].message.content

    def _create_new_tasks(self, result: str, current_task: Dict) -> List[Dict]:
        """Crée de nouvelles tâches basées sur le résultat"""

        # Ne créer de nouvelles tâches que si pertinent
        incomplete_tasks = "\n".join([
            f"{t['task_id']}. {t['task_name']}"
            for t in self.task_queue
        ])

        prompt = f"""Tu es un créateur de tâches.

Objectif : {self.objective}

Dernière tâche complétée : {current_task['task_name']}
Résultat : {result}

Tâches restantes :
{incomplete_tasks if incomplete_tasks else "Aucune"}

Basé sur le résultat, crée de nouvelles tâches pour progresser vers l'objectif.
Ne crée des tâches que si nécessaire (pas de duplication).

Réponds en JSON : {{"tasks": ["tâche 1", "tâche 2", ...]}} ou {{"tasks": []}} si aucune nouvelle tâche nécessaire."""

        response = openai.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )

        try:
            data = json.loads(response.choices[0].message.content)
            task_names = data.get("tasks", [])

            new_tasks = []
            for name in task_names:
                new_tasks.append({
                    "task_id": self.next_task_id,
                    "task_name": name
                })
                self.next_task_id += 1

            return new_tasks

        except json.JSONDecodeError:
            return []

    def _prioritize_tasks(self):
        """Réorganise les tâches par priorité"""

        if not self.task_queue:
            return

        # Convertir en liste pour manipulation
        tasks = list(self.task_queue)

        tasks_text = "\n".join([
            f"{t['task_id']}. {t['task_name']}"
            for t in tasks
        ])

        prompt = f"""Tu es un prioriseur de tâches.

Objectif : {self.objective}

Tâches à prioriser :
{tasks_text}

Réordonne ces tâches par ordre de priorité (du plus important au moins important).
Réponds avec les IDs séparés par des virgules : 5,2,8,1,3"""

        response = openai.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}]
        )

        try:
            # Parser l'ordre
            order_str = response.choices[0].message.content.strip()
            ordered_ids = [int(x.strip()) for x in order_str.split(",")]

            # Réorganiser
            ordered_tasks = []
            for task_id in ordered_ids:
                task = next((t for t in tasks if t["task_id"] == task_id), None)
                if task:
                    ordered_tasks.append(task)

            # Ajouter les tâches manquées (au cas où)
            for task in tasks:
                if task not in ordered_tasks:
                    ordered_tasks.append(task)

            self.task_queue = deque(ordered_tasks)

        except Exception as e:
            print(f"⚠️ Erreur de priorisation : {e}")
            # Garder l'ordre actuel

    def _print_summary(self):
        """Affiche un résumé"""
        print(f"\n{'='*60}")
        print(f"📊 RÉSUMÉ")
        print(f"{'='*60}")
        print(f"Objectif : {self.objective}")
        print(f"Tâches complétées : {len(self.completed_tasks)}")
        print(f"Tâches restantes : {len(self.task_queue)}")

        print("\n✅ Tâches complétées :")
        for task in self.completed_tasks:
            print(f"   {task['task_id']}. {task['task_name']}")


# ========== EXEMPLE ==========

agent = BabyAGI(
    objective="Créer un plan de marketing pour un nouveau produit SaaS",
    initial_task="Identifier le public cible"
)

agent.run(max_iterations=15)

# Architecture simplifiée d'AgentGPT

class AgentGPT:
    """Agent avec interface web"""

    def __init__(self, name: str, goal: str, session_id: str):
        self.name = name
        self.goal = goal
        self.session_id = session_id

        self.state = "idle"  # idle, running, paused, completed
        self.tasks = []
        self.logs = []

    def start(self):
        """Démarre l'agent"""
        self.state = "running"
        self._emit_event("agent_started", {"name": self.name, "goal": self.goal})

        # Boucle principale (asynchrone)
        asyncio.create_task(self._run_loop())

    async def _run_loop(self):
        """Boucle asynchrone"""
        while self.state == "running":
            # Exécuter la prochaine tâche
            await self._execute_next_task()

            # Émettre l'état
            self._emit_event("task_completed", {
                "tasks_completed": len(self.completed_tasks),
                "tasks_remaining": len(self.task_queue)
            })

            # Petit délai pour éviter rate limits
            await asyncio.sleep(2)

    def pause(self):
        """Met en pause"""
        self.state = "paused"
        self._emit_event("agent_paused", {})

    def resume(self):
        """Reprend"""
        self.state = "running"
        asyncio.create_task(self._run_loop())

    def stop(self):
        """Arrête définitivement"""
        self.state = "completed"
        self._emit_event("agent_stopped", {})

    def _emit_event(self, event_type: str, data: dict):
        """Émet un événement vers le frontend"""
        # Via WebSocket ou Server-Sent Events
        event = {
            "type": event_type,
            "data": data,
            "timestamp": datetime.now().isoformat()
        }
        # websocket.send(json.dumps(event))

# Simulation de coûts

def estimate_cost(iterations: int, avg_tokens_per_call: int = 2000):
    """Estime le coût d'un agent autonome"""

    # Hypothèses :
    # - 3 appels LLM par itération (plan, execute, evaluate)
    # - GPT-4 : $0.03/1k input tokens, $0.06/1k output tokens

    total_tokens = iterations * 3 * avg_tokens_per_call
    cost = (total_tokens / 1000) * 0.045  # Mix input/output

    return {
        "iterations": iterations,
        "total_llm_calls": iterations * 3,
        "total_tokens": total_tokens,
        "estimated_cost_usd": cost
    }

# Exemples
print(estimate_cost(10))   # ~$2.70 (tâche simple)
print(estimate_cost(50))   # ~$13.50 (tâche moyenne)
print(estimate_cost(200))  # ~$54.00 (tâche complexe)

# Sur une journée complète (monitoring 24h) :
print(estimate_cost(1000)) # ~$270 par jour ! 💸

# Exemple de dérive

Objectif initial : "Recherche les meilleurs restaurants à Paris"

Dérive progressive :
1. Recherche restaurants → OK
2. Analyse avis clients → OK
3. Recherche biographie des chefs → ⚠️ Drift
4. Analyse tendances culinaires mondiales → ❌ Hors sujet
5. Recherche histoire de la cuisine française → ❌ Très hors sujet

def check_alignment(objective: str, current_task: str) -> float:
    """Vérifie l'alignement tâche-objectif (0-1)"""

    prompt = f"""Sur une échelle de 0 à 1, à quel point cette tâche est alignée avec l'objectif ?

Objectif : {objective}
Tâche : {current_task}

Réponds uniquement avec un chiffre entre 0 et 1."""

    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )

    try:
        score = float(response.choices[0].message.content.strip())
        return score
    except:
        return 0.5  # Score par défaut

# Utilisation
if check_alignment(objective, current_task) < 0.5:
    print("⚠️ DRIFT DÉTECTÉ : Tâche peu alignée avec l'objectif")

# Exemple de boucle

Tâche : "Chercher plus d'informations"
→ Exécution : "J'ai trouvé des infos, mais pas assez"
→ Nouvelle tâche : "Chercher encore plus d'informations"
→ Exécution : "J'ai trouvé des infos, mais pas assez"
→ Nouvelle tâche : "Chercher encore plus d'informations"
[BOUCLE INFINIE]

def detect_loop(task_history: List[str], window_size: int = 3) -> bool:
    """Détecte des tâches répétitives"""

    if len(task_history) < window_size * 2:
        return False

    recent = task_history[-window_size:]

    # Vérifier si les N dernières tâches sont similaires aux N précédentes
    from difflib import SequenceMatcher

    for i in range(len(recent)):
        older_task = task_history[-(window_size * 2) + i]
        recent_task = recent[i]

        similarity = SequenceMatcher(None, older_task, recent_task).ratio()

        if similarity > 0.8:
            return True  # Boucle probable

    return False

# Exemple d'amplification d'hallucination

Tâche 1 : "Recherche le chiffre d'affaires de TechCorp"
→ Hallucination : "100M€" (en réalité : 50M€)

Tâche 2 : "Calcule la croissance annuelle"
→ Basé sur la donnée hallucinée : "Croissance de 150%"

Tâche 3 : "Prédis le CA dans 2 ans"
→ Basé sur les 2 erreurs précédentes : "3,2 milliards €" (absurde)

# Exemples d'actions risquées (SANS PROTECTION)

- DELETE_FILE("/important/data.db")
- SEND_EMAIL(to="[email protected]", subject="Spam")
- API_CALL(endpoint="https://api.payment.com/charge", amount=10000)
- EXECUTE_CODE("import os; os.system('rm -rf /')")

class SafeAgent(AutoGPT):
    """Agent avec contraintes de sécurité"""

    DANGEROUS_ACTIONS = [
        "delete_file",
        "execute_code",
        "send_email",
        "make_payment"
    ]

    REQUIRES_APPROVAL = [
        "api_call",
        "database_write",
        "send_message"
    ]

    def _execute_task(self, task: str) -> str:
        # Détecter l'action
        action = self._extract_action(task)

        # Bloquer les actions dangereuses
        if action in self.DANGEROUS_ACTIONS:
            return "❌ Action bloquée : trop dangereuse"

        # Demander approbation pour actions sensibles
        if action in self.REQUIRES_APPROVAL:
            if not self._get_human_approval(task):
                return "❌ Action refusée par l'utilisateur"

        # Exécuter
        return super()._execute_task(task)

    def _get_human_approval(self, task: str) -> bool:
        """Demande l'approbation humaine"""
        print(f"\n⚠️ APPROBATION REQUISE")
        print(f"Tâche : {task}")
        response = input("Approuver ? (y/n) : ")
        return response.lower() == "y"

research_agent = AutoGPT(
    goal="Surveiller les actualités en cybersécurité et créer un rapport hebdomadaire",
    max_iterations=100,
    budget_dollars=20.0
)

# S'exécute automatiquement chaque semaine
# Coût : ~15€/semaine

content_agent = BabyAGI(
    objective="Créer un cours complet sur les transformers en IA",
    initial_task="Créer le plan du cours"
)

# Génère : plan → chapitres → exemples → exercices → quiz

code_explorer = AutoGPT(
    goal="Analyse ce repo GitHub et documente l'architecture",
    max_iterations=50
)

# Clone → Explore → Analyse → Génère documentation

agent = AutoGPT(
    goal="...",
    max_iterations=20,        # Limite d'itérations
    budget_dollars=5.0,       # Limite de coût
    timeout_minutes=30        # Limite de temps
)

class MonitoredAutoGPT(AutoGPT):
    """Agent avec monitoring"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.metrics = {
            "start_time": None,
            "iterations": 0,
            "llm_calls": 0,
            "errors": 0
        }

    def run(self):
        self.metrics["start_time"] = datetime.now()

        # Dashboard en temps réel
        dashboard_thread = threading.Thread(target=self._display_dashboard)
        dashboard_thread.daemon = True
        dashboard_thread.start()

        return super().run()

    def _display_dashboard(self):
        """Affiche un dashboard en temps réel"""
        while True:
            os.system('clear')
            print(f"{'='*60}")
            print(f"🎯 Objectif : {self.goal}")
            print(f"⏱️ Durée : {(datetime.now() - self.metrics['start_time']).seconds}s")
            print(f"🔄 Itérations : {self.metrics['iterations']}")
            print(f"💸 Coût : ${self.spent:.2f} / ${self.budget_dollars}")
            print(f"📋 Tâches : {len(self.completed_tasks)} ✅ | {len(self.task_list)} ⏳")
            print(f"{'='*60}")
            time.sleep(1)

if task_is_critical(current_task):
    print(f"⚠️ Décision critique : {current_task}")
    print("Options :")
    print("1. Continuer")
    print("2. Modifier la tâche")
    print("3. Arrêter")

    choice = input("Choix : ")

    if choice == "3":
        agent.stop()