ReAct et Patterns d'Agents IA

┌─────────────────────────────────────────────────┐
│              BOUCLE ReAct                       │
├─────────────────────────────────────────────────┤
│                                                 │
│  💭 THOUGHT  →  🔧 ACTION  →  👁️ OBSERVATION   │
│       ↑                               ↓         │
│       └───────────────────────────────┘         │
│                  (Répéter)                      │
│                                                 │
└─────────────────────────────────────────────────┘

LLM: La capitale du Japon est Tokyo, qui compte environ 14 millions d'habitants.

Thought 1: Je dois d'abord identifier la capitale du Japon.
Action 1: search("capitale du Japon")
Observation 1: La capitale du Japon est Tokyo.

Thought 2: Maintenant je dois trouver la population actuelle de Tokyo.
Action 2: search("population Tokyo 2024")
Observation 2: L'aire métropolitaine de Tokyo compte 37,4 millions d'habitants en 2024.

Thought 3: J'ai maintenant la réponse exacte.
Answer: Environ 37,4 millions d'habitants vivent dans la capitale du Japon (Tokyo).

import openai
import json
from typing import Dict, List, Any, Optional
from dataclasses import dataclass

@dataclass
class Step:
    """Une étape du raisonnement ReAct"""
    thought: str
    action: Optional[str] = None
    action_input: Optional[Dict] = None
    observation: Optional[str] = None

class ReActAgent:
    """Agent utilisant le pattern ReAct"""

    def __init__(
        self,
        llm_model: str = "gpt-4",
        tools: Dict[str, callable] = None,
        max_iterations: int = 10,
        verbose: bool = True
    ):
        self.model = llm_model
        self.tools = tools or {}
        self.max_iterations = max_iterations
        self.verbose = verbose

    def run(self, query: str) -> str:
        """Exécute l'agent avec le pattern ReAct"""

        steps: List[Step] = []
        context = self._build_initial_context(query)

        for iteration in range(self.max_iterations):
            if self.verbose:
                print(f"\n{'='*60}")
                print(f"🔄 Itération {iteration + 1}/{self.max_iterations}")
                print(f"{'='*60}")

            # 1. REASONING : Générer pensée et action
            response = self._generate_step(context, steps)

            # Parser la réponse
            thought = response.get("thought", "")
            action_name = response.get("action")
            action_input = response.get("action_input", {})

            if self.verbose:
                print(f"\n💭 Pensée : {thought}")

            # Vérifier si l'agent a terminé
            if action_name == "finish" or action_name is None:
                final_answer = response.get("final_answer", thought)
                if self.verbose:
                    print(f"\n✅ Réponse finale : {final_answer}")
                return final_answer

            # 2. ACTION : Exécuter l'outil
            if action_name not in self.tools:
                observation = f"❌ Erreur : l'outil '{action_name}' n'existe pas. Outils disponibles : {list(self.tools.keys())}"
            else:
                if self.verbose:
                    print(f"\n🔧 Action : {action_name}({action_input})")

                try:
                    observation = self.tools[action_name](**action_input)
                except Exception as e:
                    observation = f"❌ Erreur lors de l'exécution : {str(e)}"

            if self.verbose:
                print(f"👁️  Observation : {observation}")

            # 3. OBSERVATION : Enregistrer le résultat
            step = Step(
                thought=thought,
                action=action_name,
                action_input=action_input,
                observation=observation
            )
            steps.append(step)

            # Mettre à jour le contexte
            context = self._update_context(context, step)

        return "⚠️ Limite d'itérations atteinte sans réponse définitive."

    def _build_initial_context(self, query: str) -> str:
        """Construit le contexte initial avec le prompt ReAct"""
        tools_description = self._format_tools()

        prompt = f"""Tu es un agent de résolution de problèmes utilisant le pattern ReAct.

Outils disponibles :
{tools_description}

Pour chaque étape, tu dois raisonner en suivant ce format EXACT :

Thought: [ton raisonnement sur la situation actuelle]
Action: [nom de l'outil à utiliser, ou "finish" si tu as la réponse finale]
Action Input: {{"param1": "value1", "param2": "value2"}}

Quand tu as la réponse finale, utilise :
Thought: [explication finale]
Action: finish
Final Answer: [ta réponse complète]

Question : {query}

Commence maintenant."""

        return prompt

    def _format_tools(self) -> str:
        """Formate la description des outils pour le prompt"""
        descriptions = []
        for tool_name, tool_fn in self.tools.items():
            # Extraire la docstring
            doc = tool_fn.__doc__ or "Pas de description"
            descriptions.append(f"- {tool_name}: {doc.strip()}")
        return "\n".join(descriptions)

    def _generate_step(self, context: str, steps: List[Step]) -> Dict[str, Any]:
        """Génère la prochaine pensée et action via le LLM"""

        # Construire l'historique
        messages = [{"role": "user", "content": context}]

        # Ajouter les étapes précédentes
        for step in steps:
            step_text = f"Thought: {step.thought}\nAction: {step.action}\nAction Input: {json.dumps(step.action_input)}\nObservation: {step.observation}"
            messages.append({"role": "assistant", "content": step_text})

        # Demander la prochaine étape
        messages.append({"role": "user", "content": "Quelle est ta prochaine pensée et action ?"})

        response = openai.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=0.0  # Déterministe pour la reproductibilité
        )

        # Parser la réponse du LLM
        content = response.choices[0].message.content
        return self._parse_react_response(content)

    def _parse_react_response(self, content: str) -> Dict[str, Any]:
        """Parse la réponse du LLM au format ReAct"""
        result = {
            "thought": "",
            "action": None,
            "action_input": {},
            "final_answer": None
        }

        lines = content.split('\n')
        current_key = None

        for line in lines:
            line = line.strip()

            if line.startswith("Thought:"):
                result["thought"] = line.replace("Thought:", "").strip()
                current_key = "thought"

            elif line.startswith("Action:"):
                action = line.replace("Action:", "").strip()
                result["action"] = action
                current_key = "action"

            elif line.startswith("Action Input:"):
                input_str = line.replace("Action Input:", "").strip()
                try:
                    result["action_input"] = json.loads(input_str)
                except:
                    # Fallback : traiter comme un string simple
                    result["action_input"] = {"input": input_str}
                current_key = "action_input"

            elif line.startswith("Final Answer:"):
                result["final_answer"] = line.replace("Final Answer:", "").strip()
                current_key = "final_answer"

            elif current_key and line:
                # Continuer la valeur actuelle (multi-lignes)
                result[current_key] += " " + line

        return result

    def _update_context(self, context: str, step: Step) -> str:
        """Met à jour le contexte avec la dernière observation"""
        # Dans une implémentation réelle, gérer la limite de tokens
        return context


# ============= OUTILS D'EXEMPLE =============

def search_wikipedia(query: str) -> str:
    """Recherche sur Wikipédia français et retourne un résumé."""
    import wikipedia
    wikipedia.set_lang("fr")

    try:
        page = wikipedia.page(query, auto_suggest=True)
        return page.summary[:500] + "..."
    except wikipedia.DisambiguationError as e:
        return f"Plusieurs résultats : {', '.join(e.options[:3])}"
    except wikipedia.PageError:
        return f"Aucun résultat trouvé pour '{query}'"
    except Exception as e:
        return f"Erreur : {str(e)}"

def calculator(expression: str) -> str:
    """Effectue un calcul mathématique. Exemple : expression='2+2' retourne '4'."""
    try:
        result = eval(expression)  # ⚠️ Utiliser ast.literal_eval en production
        return str(result)
    except Exception as e:
        return f"Erreur de calcul : {str(e)}"

def get_current_date() -> str:
    """Retourne la date actuelle."""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d")


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

if __name__ == "__main__":
    # Créer l'agent avec des outils
    agent = ReActAgent(
        llm_model="gpt-4",
        tools={
            "search": search_wikipedia,
            "calculator": calculator,
            "get_date": get_current_date
        },
        max_iterations=10,
        verbose=True
    )

    # Exemple 1 : Question nécessitant recherche + calcul
    result = agent.run("""
    Trouve l'année de naissance de Marie Curie,
    puis calcule combien d'années se sont écoulées depuis.
    """)

    print(f"\n\n{'#'*60}")
    print(f"RÉSULTAT FINAL : {result}")
    print(f"{'#'*60}")

============================================================
🔄 Itération 1/10
============================================================

💭 Pensée : Je dois d'abord rechercher l'année de naissance de Marie Curie

🔧 Action : search({'query': 'Marie Curie'})
👁️  Observation : Marie Curie, née Maria Salomea Skłodowska le 7 novembre 1867 à Varsovie...

============================================================
🔄 Itération 2/10
============================================================

💭 Pensée : Marie Curie est née en 1867. Maintenant je dois calculer combien d'années se sont écoulées.

🔧 Action : get_date({})
👁️  Observation : 2024-11-20

============================================================
🔄 Itération 3/10
============================================================

💭 Pensée : Nous sommes en 2024. Je dois calculer 2024 - 1867

🔧 Action : calculator({'expression': '2024 - 1867'})
👁️  Observation : 157

============================================================
🔄 Itération 4/10
============================================================

✅ Réponse finale : Marie Curie est née en 1867. 157 années se sont écoulées depuis sa naissance.

############################################################
RÉSULTAT FINAL : Marie Curie est née en 1867. 157 années se sont écoulées depuis sa naissance.
############################################################

1️⃣ PLAN      : Décomposer en sous-tâches
2️⃣ EXECUTE   : Exécuter chaque tâche séquentiellement
3️⃣ SYNTHESIZE: Combiner les résultats

class PlanAndExecuteAgent:
    """Agent qui planifie puis exécute"""

    def __init__(self, llm_model="gpt-4", tools=None):
        self.model = llm_model
        self.executor = ReActAgent(llm_model=llm_model, tools=tools)

    def run(self, query: str) -> str:
        # Phase 1 : PLANIFICATION
        plan = self._create_plan(query)
        print(f"\n📋 PLAN CRÉÉ ({len(plan)} étapes) :")
        for i, step in enumerate(plan, 1):
            print(f"  {i}. {step}")

        # Phase 2 : EXÉCUTION
        results = []
        for i, step in enumerate(plan, 1):
            print(f"\n▶️  ÉTAPE {i}/{len(plan)} : {step}")
            result = self.executor.run(step)
            results.append(result)
            print(f"✅ Résultat : {result[:100]}...")

        # Phase 3 : SYNTHÈSE
        final_answer = self._synthesize(query, plan, results)
        return final_answer

    def _create_plan(self, query: str) -> List[str]:
        """Crée un plan détaillé étape par étape"""
        response = openai.chat.completions.create(
            model=self.model,
            messages=[{
                "role": "user",
                "content": f"""Décompose cette tâche en étapes simples et séquentielles.
Chaque étape doit être une question ou action spécifique.

Tâche : {query}

Réponds uniquement avec une liste numérotée :
1. [première étape]
2. [deuxième étape]
..."""
            }],
            temperature=0.0
        )

        plan_text = response.choices[0].message.content
        # Parser la liste
        steps = []
        for line in plan_text.split('\n'):
            line = line.strip()
            # Extraire les lignes qui commencent par un numéro
            if line and line[0].isdigit():
                # Retirer le numéro et le point
                step = line.split('.', 1)[1].strip()
                steps.append(step)

        return steps

    def _synthesize(self, original_query: str, plan: List[str], results: List[str]) -> str:
        """Synthétise les résultats en une réponse finale"""
        plan_results = "\n".join([
            f"{i+1}. {plan[i]}\n   → {results[i]}"
            for i in range(len(plan))
        ])

        response = openai.chat.completions.create(
            model=self.model,
            messages=[{
                "role": "user",
                "content": f"""Synthétise ces résultats en une réponse complète.

Question originale : {original_query}

Étapes et résultats :
{plan_results}

Fournis une réponse finale cohérente et complète."""
            }],
            temperature=0.0
        )

        return response.choices[0].message.content


# Exemple d'utilisation
agent = PlanAndExecuteAgent(
    llm_model="gpt-4",
    tools={
        "search": search_wikipedia,
        "calculator": calculator
    }
)

result = agent.run("""
Analyse l'impact économique de l'IA en 2024 :
- PIB généré par l'IA
- Nombre d'emplois créés vs détruits
- Top 3 des secteurs impactés
""")

📋 PLAN CRÉÉ (4 étapes) :
  1. Recherche le PIB généré par l'IA en 2024
  2. Trouve les statistiques sur les emplois créés et détruits par l'IA
  3. Identifie les 3 secteurs les plus impactés par l'IA
  4. Synthétise les informations en une analyse cohérente

▶️  ÉTAPE 1/4 : Recherche le PIB généré par l'IA en 2024
...
✅ Résultat : Selon les estimations, l'IA a généré environ 1,2 trillion de dollars...

[etc.]

1️⃣ GENERATE : Générer une réponse initiale
2️⃣ REFLECT  : Critiquer la réponse
3️⃣ IMPROVE  : Améliorer basé sur la critique
4️⃣ Répéter jusqu'à satisfaction

class ReflexionAgent:
    """Agent avec capacité d'auto-réflexion"""

    def __init__(self, llm_model="gpt-4", max_reflections=3):
        self.model = llm_model
        self.max_reflections = max_reflections

    def run(self, query: str) -> str:
        # Génération initiale
        current_answer = self._generate_initial_answer(query)
        print(f"📝 Réponse initiale : {current_answer[:200]}...")

        # Boucle de réflexion
        for i in range(self.max_reflections):
            print(f"\n🔍 RÉFLEXION {i+1}/{self.max_reflections}")

            # Auto-critique
            critique = self._reflect(query, current_answer)
            print(f"💭 Critique : {critique['feedback']}")

            if critique["is_satisfactory"]:
                print(f"✅ Réponse satisfaisante après {i+1} réflexions")
                break

            # Amélioration
            improved = self._improve(query, current_answer, critique["feedback"])
            print(f"✨ Amélioration : {improved[:200]}...")
            current_answer = improved

        return current_answer

    def _generate_initial_answer(self, query: str) -> str:
        """Génère une première réponse"""
        response = openai.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": query}]
        )
        return response.choices[0].message.content

    def _reflect(self, query: str, answer: str) -> Dict:
        """Critique la réponse actuelle"""
        response = openai.chat.completions.create(
            model=self.model,
            messages=[{
                "role": "user",
                "content": f"""Tu es un critique expert. Évalue cette réponse :

Question : {query}
Réponse : {answer}

Critères d'évaluation :
1. Exactitude factuelle
2. Complétude (tous les aspects couverts)
3. Clarté et structure
4. Exemples et preuves

Réponds en JSON :
{{
    "is_satisfactory": true/false,
    "feedback": "Critique détaillée avec suggestions d'amélioration"
}}"""
            }],
            response_format={"type": "json_object"}
        )

        return json.loads(response.choices[0].message.content)

    def _improve(self, query: str, current: str, feedback: str) -> str:
        """Améliore la réponse basée sur le feedback"""
        response = openai.chat.completions.create(
            model=self.model,
            messages=[{
                "role": "user",
                "content": f"""Améliore cette réponse en tenant compte du feedback.

Question : {query}
Réponse actuelle : {current}
Feedback : {feedback}

Fournis une version améliorée qui adresse tous les points du feedback."""
            }]
        )

        return response.choices[0].message.content


# Exemple
agent = ReflexionAgent(llm_model="gpt-4", max_reflections=3)
result = agent.run("Explique comment fonctionne un transformeur en IA")

                    Question
                       |
          ┌────────────┼────────────┐
          ↓            ↓            ↓
       Pensée A     Pensée B     Pensée C
          |            |            |
    ┌─────┼─────┐      |       ┌────┼────┐
    ↓     ↓     ↓      ↓       ↓    ↓    ↓
   A1    A2    A3     B1      C1   C2   C3

    → Évaluer chaque branche
    → Garder les meilleures
    → Continuer jusqu'à solution

class TreeOfThoughtsAgent:
    """Agent utilisant Tree of Thoughts"""

    def __init__(self, llm_model="gpt-4", branches=3, depth=3):
        self.model = llm_model
        self.branches = branches  # Nombre de pensées à générer
        self.depth = depth        # Profondeur de l'arbre

    def run(self, query: str) -> str:
        # Générer plusieurs chemins de raisonnement
        paths = self._explore(query, current_depth=0)

        # Évaluer tous les chemins
        best_path = self._select_best_path(paths)

        # Synthétiser la réponse finale
        return self._synthesize_answer(best_path)

    def _explore(self, state: str, current_depth: int, path: List = None) -> List:
        """Explore l'arbre de pensées"""
        if path is None:
            path = []

        if current_depth >= self.depth:
            return [path]

        # Générer N pensées alternatives
        thoughts = self._generate_thoughts(state, n=self.branches)

        all_paths = []
        for thought in thoughts:
            new_path = path + [thought]

            # Évaluer si cette pensée est prometteuse
            score = self._evaluate_thought(thought)

            if score > 0.3:  # Seuil de pruning
                # Continuer l'exploration
                sub_paths = self._explore(thought, current_depth + 1, new_path)
                all_paths.extend(sub_paths)

        return all_paths

    def _generate_thoughts(self, state: str, n: int) -> List[str]:
        """Génère N pensées alternatives"""
        response = openai.chat.completions.create(
            model=self.model,
            messages=[{
                "role": "user",
                "content": f"""Génère {n} approches différentes pour résoudre ceci :

{state}

Réponds avec {n} pensées numérotées."""
            }],
            n=n,  # Générer N completions
            temperature=0.8  # Plus de diversité
        )

        thoughts = [choice.message.content for choice in response.choices]
        return thoughts

    def _evaluate_thought(self, thought: str) -> float:
        """Évalue la qualité d'une pensée (0-1)"""
        response = openai.chat.completions.create(
            model=self.model,
            messages=[{
                "role": "user",
                "content": f"""Évalue cette pensée sur 10 :

{thought}

Réponds uniquement avec un chiffre entre 0 et 10."""
            }]
        )

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

    def _select_best_path(self, paths: List[List]) -> List:
        """Sélectionne le meilleur chemin"""
        # Simplification : retourner le chemin le plus long
        return max(paths, key=len)

    def _synthesize_answer(self, path: List) -> str:
        """Synthétise la réponse finale depuis le chemin"""
        reasoning_chain = "\n".join([f"Étape {i+1}: {thought}" for i, thought in enumerate(path)])

        response = openai.chat.completions.create(
            model=self.model,
            messages=[{
                "role": "user",
                "content": f"""Basé sur ce raisonnement, fournis la réponse finale :

{reasoning_chain}"""
            }]
        )

        return response.choices[0].message.content

FEW_SHOT_EXAMPLES = """
Exemple 1 :
Question: Quelle est la population de Tokyo ?
Thought: Je dois rechercher la population actuelle de Tokyo.
Action: search
Action Input: {"query": "population Tokyo 2024"}
Observation: Tokyo compte 37,4 millions d'habitants (aire métropolitaine).
Thought: J'ai trouvé l'information.
Action: finish
Final Answer: Tokyo compte environ 37,4 millions d'habitants.

Exemple 2 :
Question: Combien font 15% de 240 ?
Thought: C'est un calcul simple, je vais utiliser la calculatrice.
Action: calculator
Action Input: {"expression": "240 * 0.15"}
Observation: 36.0
Thought: Le résultat est 36.
Action: finish
Final Answer: 15% de 240 = 36.

Maintenant, résous cette question :
"""

# Ajouter au prompt initial
prompt = FEW_SHOT_EXAMPLES + f"\nQuestion: {user_query}"

def run_with_self_consistency(agent, query, n=5):
    """Exécute l'agent N fois et prend la réponse majoritaire"""
    answers = []

    for i in range(n):
        answer = agent.run(query)
        answers.append(answer)

    # Vote majoritaire
    from collections import Counter
    most_common = Counter(answers).most_common(1)[0]

    return {
        "answer": most_common[0],
        "confidence": most_common[1] / n,
        "all_answers": answers
    }

# Utilisation
result = run_with_self_consistency(agent, "Question difficile", n=5)
print(f"Réponse : {result['answer']} (confiance: {result['confidence']*100}%)")

from sentence_transformers import SentenceTransformer
import numpy as np

class DynamicFewShotAgent(ReActAgent):
    """Agent avec exemples few-shot dynamiques"""

    def __init__(self, *args, example_bank=None, **kwargs):
        super().__init__(*args, **kwargs)
        self.example_bank = example_bank or []
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')

        # Encoder tous les exemples
        self.example_embeddings = self.encoder.encode([ex["query"] for ex in self.example_bank])

    def _build_initial_context(self, query: str) -> str:
        """Surcharge pour ajouter des exemples pertinents"""
        # Trouver les 3 exemples les plus similaires
        query_embedding = self.encoder.encode([query])
        similarities = np.dot(self.example_embeddings, query_embedding.T).flatten()
        top_indices = similarities.argsort()[-3:][::-1]

        # Construire le prompt avec les exemples
        examples = "\n\n".join([self._format_example(self.example_bank[i]) for i in top_indices])

        return f"""{examples}

Maintenant, résous cette question :
{query}"""

    def _format_example(self, example: Dict) -> str:
        return f"""Question: {example['query']}
{example['trajectory']}
Final Answer: {example['answer']}"""