Architecture et composants fondamentaux des agents IA

┌─────────────────────────────────────────────────┐
│              🧠 AGENT IA                        │
├─────────────────────────────────────────────────┤
│                                                 │
│  👁️  PERCEPTION  →  🤔 RAISONNEMENT            │
│         ↑                    ↓                  │
│  💾  MÉMOIRE    ←    🛠️  ACTION                │
│                                                 │
└─────────────────────────────────────────────────┘

"""
┌──────────────────────────────────────────────────────────────┐
│                      UTILISATEUR                             │
└───────────────────────┬──────────────────────────────────────┘
                        │
                        ↓ Query
┌───────────────────────────────────────────────────────────────┐
│                   ORCHESTRATEUR                               │
│  ┌──────────────────────────────────────────────────────┐    │
│  │  1. Parse la requête                                 │    │
│  │  2. Initialise la mémoire                            │    │
│  │  3. Lance la boucle agent                            │    │
│  │  4. Gère les erreurs et timeouts                     │    │
│  └──────────────────────────────────────────────────────┘    │
└───────┬───────────────────────────────────────────────────────┘
        │
        ↓
┌───────────────────────────────────────────────────────────────┐
│                    PERCEPTION MODULE                          │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐          │
│  │ Input Parser│  │   Sensors   │  │ Tool Results│          │
│  └─────────────┘  └─────────────┘  └─────────────┘          │
└───────┬───────────────────────────────────────────────────────┘
        │
        ↓ Observations
┌───────────────────────────────────────────────────────────────┐
│                   REASONING MODULE                            │
│  ┌─────────────────────────────────────────────────────┐     │
│  │              LLM (GPT-4, Claude)                    │     │
│  │  - Analyze observations                             │     │
│  │  - Plan next steps                                  │     │
│  │  - Decide on action                                 │     │
│  └─────────────────────────────────────────────────────┘     │
└───────┬───────────────────────────────────────────────────────┘
        │
        ↓ Decision
┌───────────────────────────────────────────────────────────────┐
│                    ACTION MODULE                              │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐          │
│  │  API Calls  │  │Code Executor│  │  Database   │          │
│  └─────────────┘  └─────────────┘  └─────────────┘          │
└───────┬───────────────────────────────────────────────────────┘
        │
        ↓ Results
┌───────────────────────────────────────────────────────────────┐
│                    MEMORY MODULE                              │
│  ┌─────────────────┐       ┌──────────────────┐             │
│  │  Short-term     │       │   Long-term      │             │
│  │  (Conversation) │       │  (Vector DB)     │             │
│  └─────────────────┘       └──────────────────┘             │
└───────────────────────────────────────────────────────────────┘

from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from abc import ABC, abstractmethod
import openai

# ============= 1. PERCEPTION =============

@dataclass
class Observation:
    """Une observation de l'environnement"""
    source: str  # "user", "tool", "system"
    content: str
    metadata: Dict[str, Any] = None

class PerceptionModule:
    """Collecte et structure les observations"""

    def __init__(self):
        self.sensors = []

    def observe(self, raw_input: Any) -> Observation:
        """Transforme un input brut en observation structurée"""
        if isinstance(raw_input, str):
            return Observation(source="user", content=raw_input)
        elif isinstance(raw_input, dict):
            return Observation(
                source=raw_input.get("source", "unknown"),
                content=raw_input.get("content", ""),
                metadata=raw_input.get("metadata", {})
            )
        else:
            return Observation(source="system", content=str(raw_input))

    def add_sensor(self, sensor_fn):
        """Ajoute un capteur (fonction qui retourne des observations)"""
        self.sensors.append(sensor_fn)

    def gather_context(self) -> List[Observation]:
        """Collecte toutes les observations des capteurs"""
        observations = []
        for sensor in self.sensors:
            try:
                obs = sensor()
                observations.append(obs)
            except Exception as e:
                observations.append(
                    Observation(source="system", content=f"Sensor error: {e}")
                )
        return observations


# ============= 2. RAISONNEMENT =============

class ReasoningStrategy(ABC):
    """Interface pour les stratégies de raisonnement"""

    @abstractmethod
    def think(self, observations: List[Observation], memory: 'Memory') -> Dict[str, Any]:
        """Analyse la situation et décide de la prochaine action"""
        pass

class ReActReasoning(ReasoningStrategy):
    """Stratégie ReAct : Thought → Action → Observation"""

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

    def think(self, observations: List[Observation], memory: 'Memory') -> Dict[str, Any]:
        """
        Génère une pensée (thought) et décide d'une action
        """
        # Construire le contexte
        context = self._build_context(observations, memory)

        # Appel LLM pour raisonner
        response = openai.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": self._get_system_prompt()},
                {"role": "user", "content": context}
            ],
            functions=self._get_available_functions(),
            function_call="auto"
        )

        message = response.choices[0].message

        # Extraire la pensée et l'action
        thought = message.content or "Réflexion en cours..."

        action = None
        if message.function_call:
            action = {
                "name": message.function_call.name,
                "arguments": json.loads(message.function_call.arguments)
            }

        return {
            "thought": thought,
            "action": action,
            "finish": action is None  # Si pas d'action, c'est fini
        }

    def _build_context(self, observations: List[Observation], memory: 'Memory') -> str:
        """Construit le contexte pour le LLM"""
        context_parts = []

        # Historique de la conversation
        for msg in memory.short_term:
            context_parts.append(f"{msg['role']}: {msg['content']}")

        # Observations récentes
        context_parts.append("\n=== Observations récentes ===")
        for obs in observations[-5:]:  # Dernières 5 observations
            context_parts.append(f"[{obs.source}] {obs.content}")

        return "\n".join(context_parts)

    def _get_system_prompt(self) -> str:
        return """Tu es un agent IA qui résout des problèmes étape par étape.

Pour chaque requête :
1. PENSE : Analyse la situation
2. AGIS : Choisis un outil à utiliser (ou réponds directement)
3. OBSERVE : Analyse le résultat
4. Répète jusqu'à résolution

Sois concis et méthodique."""

    def _get_available_functions(self) -> List[Dict]:
        """Retourne la liste des fonctions disponibles pour le LLM"""
        return [
            {
                "name": "search_web",
                "description": "Recherche sur le web",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "Requête de recherche"}
                    },
                    "required": ["query"]
                }
            },
            {
                "name": "calculate",
                "description": "Effectue un calcul mathématique",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "expression": {"type": "string", "description": "Expression mathématique"}
                    },
                    "required": ["expression"]
                }
            }
        ]


# ============= 3. ACTION =============

class Tool(ABC):
    """Interface pour un outil"""

    @abstractmethod
    def execute(self, **kwargs) -> Any:
        """Exécute l'outil avec les arguments fournis"""
        pass

    @property
    @abstractmethod
    def name(self) -> str:
        """Nom de l'outil"""
        pass

class CalculatorTool(Tool):
    """Outil de calcul"""

    @property
    def name(self) -> str:
        return "calculate"

    def execute(self, expression: str) -> str:
        try:
            result = eval(expression)  # ⚠️ Dangereux en prod
            return f"Résultat : {result}"
        except Exception as e:
            return f"Erreur de calcul : {str(e)}"

class WebSearchTool(Tool):
    """Outil de recherche web (simulé)"""

    @property
    def name(self) -> str:
        return "search_web"

    def execute(self, query: str) -> str:
        # En production, utiliser une vraie API (Tavily, Serper, etc.)
        return f"Résultats de recherche pour '{query}': [Simulé] Top 3 résultats..."

class ActionModule:
    """Gère l'exécution des actions"""

    def __init__(self):
        self.tools: Dict[str, Tool] = {}

    def register_tool(self, tool: Tool):
        """Enregistre un nouvel outil"""
        self.tools[tool.name] = tool

    def execute(self, action: Dict[str, Any]) -> Observation:
        """Exécute une action et retourne l'observation résultante"""
        tool_name = action["name"]
        arguments = action["arguments"]

        if tool_name not in self.tools:
            return Observation(
                source="system",
                content=f"Outil inconnu : {tool_name}"
            )

        try:
            result = self.tools[tool_name].execute(**arguments)
            return Observation(
                source="tool",
                content=result,
                metadata={"tool": tool_name, "arguments": arguments}
            )
        except Exception as e:
            return Observation(
                source="system",
                content=f"Erreur lors de l'exécution de {tool_name} : {str(e)}"
            )


# ============= 4. MÉMOIRE =============

class Memory:
    """Gère la mémoire court-terme et long-terme"""

    def __init__(self):
        self.short_term: List[Dict[str, str]] = []  # Conversation actuelle
        self.long_term: Dict[str, Any] = {}  # Faits persistants
        self.episodic: List[Dict[str, Any]] = []  # Historique des actions

    def add_message(self, role: str, content: str):
        """Ajoute un message à la mémoire court-terme"""
        self.short_term.append({"role": role, "content": content})

    def remember_fact(self, key: str, value: Any):
        """Stocke un fait en mémoire long-terme"""
        self.long_term[key] = value

    def recall_fact(self, key: str) -> Optional[Any]:
        """Récupère un fait de la mémoire long-terme"""
        return self.long_term.get(key)

    def add_episode(self, action: str, result: str, success: bool):
        """Ajoute un épisode à la mémoire épisodique"""
        self.episodic.append({
            "action": action,
            "result": result,
            "success": success,
            "timestamp": datetime.now().isoformat()
        })

    def get_recent_episodes(self, n: int = 5) -> List[Dict]:
        """Récupère les N derniers épisodes"""
        return self.episodic[-n:]

    def clear_short_term(self):
        """Réinitialise la mémoire court-terme"""
        self.short_term = []


# ============= 5. ORCHESTRATEUR =============

class Agent:
    """Orchestrateur principal de l'agent"""

    def __init__(
        self,
        reasoning_strategy: ReasoningStrategy,
        max_iterations: int = 10
    ):
        self.perception = PerceptionModule()
        self.reasoning = reasoning_strategy
        self.action = ActionModule()
        self.memory = Memory()
        self.max_iterations = max_iterations

    def run(self, query: str) -> str:
        """
        Exécute l'agent sur une requête
        """
        # 1. Perception : Observer la requête
        initial_observation = self.perception.observe(query)
        self.memory.add_message("user", query)

        observations = [initial_observation]
        iteration = 0

        while iteration < self.max_iterations:
            print(f"\n{'='*60}")
            print(f"🔄 Itération {iteration + 1}/{self.max_iterations}")
            print(f"{'='*60}")

            # 2. Raisonnement : Analyser et décider
            decision = self.reasoning.think(observations, self.memory)

            thought = decision["thought"]
            action = decision["action"]
            finish = decision["finish"]

            print(f"\n💭 Pensée : {thought}")

            # Sauvegarder la pensée
            self.memory.add_message("assistant", f"[THOUGHT] {thought}")

            # 3. Action : Exécuter si nécessaire
            if finish:
                print(f"\n✅ Réponse finale : {thought}")
                return thought

            if action:
                print(f"\n🔧 Action : {action['name']}({action['arguments']})")

                # Exécuter l'action
                result_observation = self.action.execute(action)
                observations.append(result_observation)

                print(f"📊 Résultat : {result_observation.content}")

                # Sauvegarder dans la mémoire
                self.memory.add_message("tool", result_observation.content)
                self.memory.add_episode(
                    action=action["name"],
                    result=result_observation.content,
                    success="Erreur" not in result_observation.content
                )

            iteration += 1

        return "⚠️ Limite d'itérations atteinte"


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

if __name__ == "__main__":
    # Créer l'agent
    agent = Agent(
        reasoning_strategy=ReActReasoning(llm_model="gpt-4"),
        max_iterations=10
    )

    # Enregistrer les outils
    agent.action.register_tool(CalculatorTool())
    agent.action.register_tool(WebSearchTool())

    # Exécuter
    result = agent.run("""
    Cherche sur le web le PIB de la France en 2023,
    puis calcule combien ça fait par habitant (67 millions d'habitants)
    """)

    print(f"\n{'='*60}")
    print(f"📝 Résultat final : {result}")
    print(f"{'='*60}")

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

💭 Pensée : Je dois d'abord rechercher le PIB de la France en 2023

🔧 Action : search_web({'query': 'PIB France 2023'})
📊 Résultat : Résultats de recherche pour 'PIB France 2023': Le PIB de la France en 2023 est estimé à 2 950 milliards d'euros...

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

💭 Pensée : Maintenant je dois calculer le PIB par habitant : 2950000000000 / 67000000

🔧 Action : calculate({'expression': '2950000000000 / 67000000'})
📊 Résultat : Résultat : 44029.85

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

✅ Réponse finale : Le PIB de la France en 2023 est d'environ 2 950 milliards d'euros. Avec 67 millions d'habitants, cela fait environ 44 030 € par habitant.

class HierarchicalAgent(Agent):
    """Agent avec délégation à des sous-agents"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.sub_agents = {}

    def register_sub_agent(self, name: str, agent: Agent):
        """Enregistre un sous-agent spécialisé"""
        self.sub_agents[name] = agent

    def delegate(self, task: str, sub_agent_name: str) -> str:
        """Délègue une tâche à un sous-agent"""
        if sub_agent_name not in self.sub_agents:
            return f"Sous-agent {sub_agent_name} introuvable"

        return self.sub_agents[sub_agent_name].run(task)

# Exemple
supervisor = HierarchicalAgent(reasoning_strategy=ReActReasoning())

# Agents spécialisés
data_agent = Agent(reasoning_strategy=ReActReasoning())
data_agent.action.register_tool(SQLTool())

research_agent = Agent(reasoning_strategy=ReActReasoning())
research_agent.action.register_tool(WebSearchTool())

# Enregistrement
supervisor.register_sub_agent("data", data_agent)
supervisor.register_sub_agent("research", research_agent)

class PlanAndExecuteAgent(Agent):
    """Agent qui planifie d'abord, puis exécute"""

    def run(self, query: str) -> str:
        # 1. Planification : Décomposer en sous-tâches
        plan = self._create_plan(query)
        print(f"📋 Plan créé : {len(plan)} étapes")

        results = []

        # 2. Exécution : Traiter chaque étape
        for i, step in enumerate(plan):
            print(f"\n▶️ Étape {i+1}/{len(plan)}: {step}")
            step_result = super().run(step)
            results.append(step_result)

        # 3. Synthèse finale
        return self._synthesize(query, results)

    def _create_plan(self, query: str) -> List[str]:
        """Utilise le LLM pour créer un plan"""
        response = openai.chat.completions.create(
            model="gpt-4",
            messages=[{
                "role": "user",
                "content": f"""Décompose cette tâche en étapes simples et séquentielles :

{query}

Réponds uniquement avec une liste numérotée."""
            }]
        )

        plan_text = response.choices[0].message.content
        # Parser la liste
        steps = [line.strip() for line in plan_text.split('\n') if line.strip()]
        return steps

    def _synthesize(self, original_query: str, step_results: List[str]) -> str:
        """Synthétise les résultats de toutes les étapes"""
        synthesis_prompt = f"""Requête originale : {original_query}

Résultats des étapes :
{chr(10).join(f'{i+1}. {r}' for i, r in enumerate(step_results))}

Fournis une réponse finale synthétique."""

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

        return response.choices[0].message.content

class ReflectiveAgent(Agent):
    """Agent qui critique et améliore ses propres résultats"""

    def run(self, query: str, max_reflections: int = 3) -> str:
        current_answer = super().run(query)

        for i in range(max_reflections):
            # Auto-critique
            critique = self._critique(query, current_answer)

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

            print(f"\n🔍 Réflexion {i+1} : {critique['issues']}")

            # Amélioration
            improved_answer = self._improve(query, current_answer, critique)
            current_answer = improved_answer

        return current_answer

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

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

La réponse est-elle :
1. Correcte factuellement ?
2. Complète ?
3. Bien structurée ?

Réponds en JSON : {{"is_satisfactory": bool, "issues": str}}"""
            }],
            response_format={"type": "json_object"}
        )

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

    def _improve(self, query: str, current: str, critique: Dict) -> str:
        """Améliore la réponse basée sur la critique"""
        response = openai.chat.completions.create(
            model="gpt-4",
            messages=[{
                "role": "user",
                "content": f"""Améliore cette réponse :

Question : {query}
Réponse actuelle : {current}
Problèmes identifiés : {critique['issues']}

Fournis une réponse améliorée."""
            }]
        )

        return response.choices[0].message.content

from functools import lru_cache
import hashlib

class CachedAgent(Agent):
    """Agent avec cache des décisions récurrentes"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.decision_cache = {}

    def run(self, query: str) -> str:
        # Créer une clé de cache
        cache_key = hashlib.md5(query.encode()).hexdigest()

        if cache_key in self.decision_cache:
            print("⚡ Réponse en cache")
            return self.decision_cache[cache_key]

        result = super().run(query)
        self.decision_cache[cache_key] = result

        return result

import asyncio
from typing import List

class AsyncActionModule(ActionModule):
    """Module d'action avec exécution parallèle"""

    async def execute_parallel(self, actions: List[Dict]) -> List[Observation]:
        """Exécute plusieurs actions en parallèle"""
        tasks = [self._execute_async(action) for action in actions]
        return await asyncio.gather(*tasks)

    async def _execute_async(self, action: Dict) -> Observation:
        """Version asynchrone de execute()"""
        # Simuler une opération async
        await asyncio.sleep(0.1)
        return self.execute(action)

# Exemple d'utilisation
async def main():
    agent = Agent(reasoning_strategy=ReActReasoning())
    agent.action = AsyncActionModule()

    actions = [
        {"name": "search_web", "arguments": {"query": "météo Paris"}},
        {"name": "search_web", "arguments": {"query": "météo Lyon"}},
        {"name": "search_web", "arguments": {"query": "météo Marseille"}}
    ]

    # Exécution parallèle = 3x plus rapide
    results = await agent.action.execute_parallel(actions)

class StreamingAgent(Agent):
    """Agent qui stream ses résultats en temps réel"""

    def run_stream(self, query: str):
        """Version streaming de run()"""
        observations = [self.perception.observe(query)]

        for iteration in range(self.max_iterations):
            decision = self.reasoning.think(observations, self.memory)

            # Stream la pensée
            yield {"type": "thought", "content": decision["thought"]}

            if decision["finish"]:
                yield {"type": "final", "content": decision["thought"]}
                break

            if decision["action"]:
                # Stream l'action
                yield {"type": "action", "content": decision["action"]}

                result = self.action.execute(decision["action"])
                observations.append(result)

                # Stream le résultat
                yield {"type": "observation", "content": result.content}

# Utilisation
agent = StreamingAgent(reasoning_strategy=ReActReasoning())

for event in agent.run_stream("Quelle est la météo à Paris ?"):
    if event["type"] == "thought":
        print(f"💭 {event['content']}")
    elif event["type"] == "action":
        print(f"🔧 {event['content']}")
    elif event["type"] == "observation":
        print(f"📊 {event['content']}")
    elif event["type"] == "final":
        print(f"✅ {event['content']}")

class RobustAgent(Agent):
    """Agent avec gestion avancée des erreurs"""

    def __init__(self, *args, retry_on_error=True, **kwargs):
        super().__init__(*args, **kwargs)
        self.retry_on_error = retry_on_error
        self.error_count = 0
        self.max_errors = 3

    def run(self, query: str) -> str:
        try:
            return self._run_with_fallback(query)
        except Exception as e:
            return self._handle_critical_error(e)

    def _run_with_fallback(self, query: str) -> str:
        """Exécution avec fallback sur erreur"""
        try:
            return super().run(query)
        except ToolExecutionError as e:
            if self.retry_on_error and self.error_count < self.max_errors:
                self.error_count += 1
                print(f"⚠️ Erreur outil, retry {self.error_count}/{self.max_errors}")
                return self._run_with_fallback(query)
            else:
                # Fallback : réponse sans outils
                return self._generate_fallback_response(query)

    def _handle_critical_error(self, error: Exception) -> str:
        """Gestion des erreurs critiques"""
        return f"Désolé, une erreur s'est produite : {str(error)}. Veuillez réessayer."

    def _generate_fallback_response(self, query: str) -> str:
        """Génère une réponse sans utiliser d'outils"""
        response = openai.chat.completions.create(
            model="gpt-4",
            messages=[{
                "role": "user",
                "content": f"{query}\n\nRéponds sans utiliser d'outils externes."
            }]
        )
        return response.choices[0].message.content

from dataclasses import dataclass
from datetime import datetime
import time

@dataclass
class AgentMetrics:
    """Métriques d'exécution d'un agent"""
    iterations: int
    tools_called: int
    total_duration_s: float
    llm_calls: int
    tokens_used: int
    success: bool
    error_message: Optional[str] = None

class ObservableAgent(Agent):
    """Agent avec collecte de métriques"""

    def run(self, query: str) -> str:
        start_time = time.time()
        metrics = AgentMetrics(
            iterations=0,
            tools_called=0,
            total_duration_s=0.0,
            llm_calls=0,
            tokens_used=0,
            success=False
        )

        try:
            result = super().run(query)
            metrics.success = True
            return result
        except Exception as e:
            metrics.error_message = str(e)
            raise
        finally:
            metrics.total_duration_s = time.time() - start_time
            self._log_metrics(metrics)

    def _log_metrics(self, metrics: AgentMetrics):
        """Log les métriques (vers un système de monitoring)"""
        print(f"\n📊 Métriques d'exécution:")
        print(f"  - Itérations : {metrics.iterations}")
        print(f"  - Outils appelés : {metrics.tools_called}")
        print(f"  - Durée : {metrics.total_duration_s:.2f}s")
        print(f"  - Appels LLM : {metrics.llm_calls}")
        print(f"  - Succès : {metrics.success}")

class TranslationTool(Tool):
    @property
    def name(self) -> str:
        return "translate"

    def execute(self, text: str, target_language: str) -> str:
        # TODO: Implémenter
        pass

# Timeout par outil (3 secondes max)
@timeout(3)
def execute_tool(tool, args):
    return tool.execute(**args)

# Timeout global de l'agent (30 secondes max)
@timeout(30)
def agent.run(query):
    # ...

class SafeTool(Tool):
    def execute(self, **kwargs):
        # 1. Valider entrées
        self._validate_inputs(kwargs)

        # 2. Exécuter
        result = self._execute(**kwargs)

        # 3. Valider sortie
        self._validate_output(result)

        return result

class RateLimitedAgent(Agent):
    def __init__(self, *args, max_cost_usd=0.50, **kwargs):
        super().__init__(*args, **kwargs)
        self.max_cost = max_cost_usd
        self.current_cost = 0.0

    def run(self, query):
        if self.current_cost >= self.max_cost:
            raise BudgetExceededError(
                f"Budget dépassé : ${self.current_cost:.2f}"
            )
        return super().run(query)

import uuid
import logging

class TracedAgent(Agent):
    def run(self, query):
        trace_id = str(uuid.uuid4())
        logger = logging.getLogger(f"agent.{trace_id}")

        logger.info(f"[START] Query: {query}")

        try:
            result = super().run(query)
            logger.info(f"[SUCCESS] Result: {result}")
            return result
        except Exception as e:
            logger.error(f"[ERROR] {str(e)}", exc_info=True)
            raise