Les 5 patterns d'agents IA qui marchent

Question de l'utilisateur
         │
         ▼
    ┌─────────┐
    │ THOUGHT  │ ← "Je dois chercher la population de la France"
    └────┬────┘
         │
         ▼
    ┌─────────┐
    │  ACTION  │ ← search("population France 2024")
    └────┬────┘
         │
         ▼
    ┌─────────┐
    │OBSERVAT° │ ← "68,4 millions d'habitants"
    └────┬────┘
         │
         ▼
    ┌─────────┐
    │ THOUGHT  │ ← "J'ai l'info, je peux répondre"
    └────┬────┘
         │
         ▼
    ┌─────────┐
    │ RÉPONSE  │ ← "La France compte 68,4 millions d'habitants"
    └─────────┘

class ReActAgent:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
        self.max_iterations = 10

    def run(self, question: str) -> str:
        prompt = f"""Tu es un agent ReAct. Pour chaque étape :
Thought: ton raisonnement
Action: tool_name(params)
Observation: (sera rempli automatiquement)
...
Quand tu as la réponse finale :
Thought: j'ai assez d'informations
Final Answer: ta réponse

Question: {question}"""

        history = prompt
        for i in range(self.max_iterations):
            response = self.llm.complete(history)
            history += response

            if "Final Answer:" in response:
                return response.split("Final Answer:")[-1].strip()

            if "Action:" in response:
                action = self.parse_action(response)
                result = self.execute_tool(action)
                observation = f"\nObservation: {result}\n"
                history += observation

        return "Désolé, je n'ai pas pu trouver la réponse."

    def parse_action(self, text):
        # Extraire le nom de l'outil et les paramètres
        action_line = [l for l in text.split("\n") if l.startswith("Action:")][0]
        return self.parse_tool_call(action_line)

    def execute_tool(self, action):
        tool = self.tools.get(action["name"])
        if tool:
            return tool(**action["params"])
        return f"Outil '{action['name']}' non trouvé"

Question de l'utilisateur
         │
         ▼
┌──────────────────┐
│    PLANIFICATEUR   │
│  (LLM puissant)   │
│                    │
│  Plan:             │
│  1. Chercher X     │
│  2. Analyser Y     │
│  3. Calculer Z     │
│  4. Synthétiser    │
└────────┬───────────┘
         │
         ▼
┌──────────────────┐
│    EXÉCUTEUR       │     ┌──────────┐
│  (LLM + outils)   │────▶│ Étape 1  │ ✅
│                    │     └──────────┘
│                    │     ┌──────────┐
│                    │────▶│ Étape 2  │ ✅
│                    │     └──────────┘
│                    │     ┌──────────┐
│                    │────▶│ Étape 3  │ ❌ Échec !
│                    │     └──────────┘
└────────┬───────────┘
         │
         ▼
┌──────────────────┐
│   RE-PLANIFICATEUR │
│   Ajuste le plan   │
│   3bis. Alternative│
│   4. Synthétiser   │
└──────────────────┘

class PlanAndExecuteAgent:
    def __init__(self, planner_llm, executor_llm, tools):
        self.planner = planner_llm   # Modèle puissant (ex: GPT-4, Claude Opus)
        self.executor = executor_llm  # Modèle rapide (ex: GPT-4o-mini, Haiku)
        self.tools = tools

    def run(self, task: str) -> str:
        # Phase 1 : Planification
        plan = self.create_plan(task)
        results = []

        # Phase 2 : Exécution
        for i, step in enumerate(plan):
            try:
                result = self.execute_step(step, results)
                results.append({"step": step, "result": result, "status": "success"})
            except Exception as e:
                results.append({"step": step, "result": str(e), "status": "failed"})
                # Re-planifier si nécessaire
                plan = self.replan(task, plan, results, i)

        # Phase 3 : Synthèse
        return self.synthesize(task, results)

    def create_plan(self, task: str) -> list:
        prompt = f"""Crée un plan étape par étape pour accomplir cette tâche.
Retourne une liste JSON d'étapes.

Tâche: {task}

Outils disponibles: {list(self.tools.keys())}"""

        response = self.planner.complete(prompt)
        return json.loads(response)

    def execute_step(self, step: dict, previous_results: list) -> str:
        context = "\n".join(
            f"Étape {i+1}: {r['result']}"
            for i, r in enumerate(previous_results)
        )
        prompt = f"""Exécute cette étape en utilisant les outils disponibles.

Contexte des étapes précédentes:
{context}

Étape à exécuter: {step['description']}"""

        return self.executor.complete_with_tools(prompt, self.tools)

    def replan(self, task, original_plan, results, failed_index):
        prompt = f"""Le plan original a échoué à l'étape {failed_index + 1}.

Plan original: {json.dumps(original_plan)}
Résultats: {json.dumps(results)}

Crée un nouveau plan à partir de l'étape {failed_index + 1}."""

        new_steps = json.loads(self.planner.complete(prompt))
        return original_plan[:failed_index] + new_steps

    def synthesize(self, task, results):
        prompt = f"""Synthétise les résultats pour répondre à la tâche.
Tâche: {task}
Résultats: {json.dumps(results)}"""
        return self.planner.complete(prompt)

Tâche de l'utilisateur
         │
         ▼
    ┌─────────┐
    │  ACTEUR  │ ← Produit une première réponse/solution
    └────┬────┘
         │
         ▼
    ┌─────────┐
    │ÉVALUATEUR│ ← Analyse la qualité (tests, critères, score)
    └────┬────┘
         │
    ┌────▼────┐
    │ Score OK │──── OUI ───▶ Réponse finale ✅
    │    ?     │
    └────┬────┘
         │ NON
         ▼
    ┌──────────┐
    │ RÉFLEXION │ ← "La réponse manque X, l'erreur est Y"
    └────┬─────┘
         │
         ▼
    ┌─────────┐
    │  ACTEUR  │ ← Nouvelle tentative avec la réflexion en contexte
    └────┬────┘
         │
         ▼
       (boucle)

class ReflexionAgent:
    def __init__(self, llm, evaluator, max_retries=3):
        self.llm = llm
        self.evaluator = evaluator
        self.max_retries = max_retries

    def run(self, task: str) -> str:
        reflections = []

        for attempt in range(self.max_retries + 1):
            # Phase 1 : Produire une réponse
            response = self.act(task, reflections)

            # Phase 2 : Évaluer
            evaluation = self.evaluate(task, response)

            if evaluation["passed"]:
                return response

            # Phase 3 : Réfléchir
            reflection = self.reflect(task, response, evaluation)
            reflections.append(reflection)

        # Retourner la meilleure tentative
        return response

    def act(self, task: str, reflections: list) -> str:
        reflection_context = ""
        if reflections:
            reflection_context = "\n\nRéflexions sur tes tentatives précédentes:\n"
            for i, r in enumerate(reflections):
                reflection_context += f"\nTentative {i+1}: {r}\n"

        prompt = f"""Accomplis cette tâche du mieux possible.
{reflection_context}
Tâche: {task}"""
        return self.llm.complete(prompt)

    def evaluate(self, task: str, response: str) -> dict:
        # Peut être un LLM, des tests automatisés, ou un score
        prompt = f"""Évalue cette réponse sur une échelle de 1-10.
Identifie les problèmes spécifiques.

Tâche: {task}
Réponse: {response}

Retourne un JSON: {{"score": X, "passed": bool, "issues": [...]}}"""

        result = self.evaluator.complete(prompt)
        return json.loads(result)

    def reflect(self, task, response, evaluation) -> str:
        prompt = f"""Analyse pourquoi cette réponse n'est pas satisfaisante
et donne des instructions précises pour l'améliorer.

Tâche: {task}
Réponse: {response}
Problèmes: {evaluation['issues']}

Réflexion:"""
        return self.llm.complete(prompt)

                    Tâche de l'utilisateur
                             │
                             ▼
                    ┌─────────────────┐
                    │   ORCHESTRATEUR   │
                    │   (Coordinateur)  │
                    └────────┬────────┘
                             │
              ┌──────────────┼──────────────┐
              ▼              ▼              ▼
       ┌────────────┐ ┌────────────┐ ┌────────────┐
       │  AGENT      │ │  AGENT      │ │  AGENT      │
       │  Recherche  │ │  Analyse    │ │  Rédaction  │
       │             │ │             │ │             │
       │  Outils:    │ │  Outils:    │ │  Outils:    │
       │  - Web      │ │  - Python   │ │  - Markdown │
       │  - News API │ │  - Charts   │ │  - CMS      │
       └──────┬─────┘ └──────┬─────┘ └──────┬─────┘
              │              │              │
              └──────────────┼──────────────┘
                             │
                             ▼
                      Résultat final

class Agent:
    def __init__(self, name, role, llm, tools=None):
        self.name = name
        self.role = role
        self.llm = llm
        self.tools = tools or []

    def execute(self, task: str, context: str = "") -> str:
        prompt = f"""Tu es {self.name}, un agent spécialisé.
Ton rôle: {self.role}

Contexte des étapes précédentes:
{context}

Tâche: {task}"""
        return self.llm.complete_with_tools(prompt, self.tools)


class MultiAgentOrchestrator:
    def __init__(self):
        self.agents = {}

    def add_agent(self, agent: Agent):
        self.agents[agent.name] = agent

    def run_sequential(self, task: str, agent_order: list) -> str:
        """Exécution séquentielle : chaque agent passe son résultat au suivant."""
        context = ""
        for agent_name in agent_order:
            agent = self.agents[agent_name]
            result = agent.execute(task, context)
            context += f"\n\n[{agent.name}]:\n{result}"
        return context

    def run_hierarchical(self, task: str, manager_name: str) -> str:
        """Le manager décide quel agent appeler et quand."""
        manager = self.agents[manager_name]
        workers = {k: v for k, v in self.agents.items() if k != manager_name}

        worker_descriptions = "\n".join(
            f"- {name}: {a.role}" for name, a in workers.items()
        )

        context = ""
        for _ in range(10):  # Max 10 délégations
            prompt = f"""Tu es le manager. Voici ton équipe:
{worker_descriptions}

Tâche globale: {task}
Travail fait: {context}

Délègue la prochaine sous-tâche à un agent, ou dis TERMINÉ si tout est fait.
Format: DELEGATE:agent_name:sous-tâche OU TERMINÉ:synthèse"""

            decision = manager.llm.complete(prompt)

            if decision.startswith("TERMINÉ"):
                return decision.split(":", 1)[1]

            if decision.startswith("DELEGATE"):
                _, agent_name, subtask = decision.split(":", 2)
                agent = workers.get(agent_name.strip())
                if agent:
                    result = agent.execute(subtask, context)
                    context += f"\n[{agent.name}]: {result}"

        return context


# Utilisation
orchestrator = MultiAgentOrchestrator()

orchestrator.add_agent(Agent(
    "researcher", "Expert en recherche web",
    llm=fast_llm, tools=[web_search, web_fetch]
))
orchestrator.add_agent(Agent(
    "analyst", "Expert en analyse de données",
    llm=smart_llm, tools=[python_exec, chart_gen]
))
orchestrator.add_agent(Agent(
    "writer", "Rédacteur professionnel",
    llm=smart_llm, tools=[markdown_gen]
))
orchestrator.add_agent(Agent(
    "manager", "Chef de projet qui coordonne l'équipe",
    llm=smart_llm
))

# Exécution hiérarchique
result = orchestrator.run_hierarchical(
    "Rédige un rapport sur les tendances IA en 2025",
    manager_name="manager"
)

Question de l'utilisateur
         │
         ▼
    ┌─────────────┐
    │  AGENT RAG   │
    │  (LLM)       │
    └──────┬──────┘
           │
           │ Choix dynamique de la source
           │
     ┌─────┼──────┬────────────┐
     ▼     ▼      ▼            ▼
┌────────┐┌────┐┌──────┐┌──────────┐
│Vector  ││SQL ││ API  ││  Web     │
│Search  ││    ││externe││ Search   │
│        ││    ││      ││          │
│Pinecone││PG  ││REST  ││Brave/    │
│Chroma  ││    ││      ││Google    │
└───┬────┘└──┬─┘└──┬───┘└────┬────┘
    │        │     │         │
    └────────┴─────┴─────────┘
                   │
                   ▼
           ┌──────────────┐
           │  Synthèse     │
           │  avec sources │
           └──────────────┘

class ToolAugmentedRAG:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = {
            "vector_search": self.vector_search,
            "sql_query": self.sql_query,
            "web_search": self.web_search,
            "api_fetch": self.api_fetch,
        }

    def vector_search(self, query: str, collection: str, top_k: int = 5) -> list:
        """Recherche sémantique dans une base vectorielle."""
        embeddings = self.embed(query)
        results = self.vector_db.query(
            collection=collection,
            vector=embeddings,
            top_k=top_k
        )
        return [{"text": r.text, "score": r.score, "source": r.metadata} for r in results]

    def sql_query(self, query: str) -> list:
        """Exécute une requête SQL en lecture seule."""
        # Validation : lecture seule !
        if not query.strip().upper().startswith("SELECT"):
            raise ValueError("Seules les requêtes SELECT sont autorisées")
        return self.db.execute(query).fetchall()

    def web_search(self, query: str, num_results: int = 5) -> list:
        """Recherche sur le web."""
        return brave_search(query, count=num_results)

    def api_fetch(self, url: str, params: dict = None) -> dict:
        """Appel à une API externe."""
        response = requests.get(url, params=params, timeout=10)
        return response.json()

    def run(self, question: str) -> str:
        sources = []
        max_searches = 5

        for i in range(max_searches):
            prompt = f"""Tu es un agent de recherche.
Question: {question}

Sources déjà trouvées:
{json.dumps(sources, indent=2) if sources else "Aucune"}

Outils disponibles:
- vector_search(query, collection, top_k): Recherche sémantique
- sql_query(query): Requête SQL
- web_search(query, num_results): Recherche web
- api_fetch(url, params): Appel API

Tu as assez d'informations pour répondre ? Si oui, réponds FINAL:ta_réponse
Sinon, appelle un outil : TOOL:nom_outil:params_json"""

            decision = self.llm.complete(prompt)

            if decision.startswith("FINAL:"):
                return decision[6:]

            if decision.startswith("TOOL:"):
                _, tool_name, params_str = decision.split(":", 2)
                params = json.loads(params_str)
                result = self.tools[tool_name](**params)
                sources.append({
                    "tool": tool_name,
                    "params": params,
                    "result": result
                })

        # Synthèse forcée après max_searches
        return self.synthesize(question, sources)

La tâche est-elle simple (1-3 étapes) ?
├── OUI → ReAct
└── NON
    ├── La qualité doit-elle être maximale ?
    │   ├── OUI → Reflexion
    │   └── NON
    │       ├── La tâche a-t-elle des étapes prévisibles ?
    │       │   ├── OUI → Plan-and-Execute
    │       │   └── NON
    │       │       ├── Faut-il des compétences variées ?
    │       │       │   ├── OUI → Multi-Agent
    │       │       │   └── NON → ReAct
    │       └── La tâche est-elle centrée sur la recherche d'info ?
    │           └── OUI → Tool-Augmented RAG