Les 5 patterns d'agents IA qui marchent

User question
         │
         ▼
    ┌─────────┐
    │ THOUGHT  │ ← "I need to look up the population of France"
    └────┬────┘
         │
         ▼
    ┌─────────┐
    │  ACTION  │ ← search("population France 2024")
    └────┬────┘
         │
         ▼
    ┌─────────┐
    │OBSERVAT° │ ← "68.4 million inhabitants"
    └────┬────┘
         │
         ▼
    ┌─────────┐
    │ THOUGHT  │ ← "I have the info, I can respond"
    └────┬────┘
         │
         ▼
    ┌─────────┐
    │ RESPONSE │ ← "France has 68.4 million inhabitants"
    └─────────┘

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"""You are a ReAct agent. For each step:
Thought: your reasoning
Action: tool_name(params)
Observation: (will be filled automatically)
...
When you have the final answer:
Thought: I have enough information
Final Answer: your answer

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 "Sorry, I couldn't find the answer."

    def parse_action(self, text):
        # Extract tool name and parameters
        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"Tool '{action['name']}' not found"

User question
         │
         ▼
┌──────────────────┐
│     PLANNER       │
│  (powerful LLM)   │
│                    │
│  Plan:             │
│  1. Search X       │
│  2. Analyze Y      │
│  3. Calculate Z    │
│  4. Synthesize     │
└────────┬───────────┘
         │
         ▼
┌──────────────────┐
│     EXECUTOR      │     ┌──────────┐
│  (LLM + tools)    │────▶│ Step 1   │ ✅
│                    │     └──────────┘
│                    │     ┌──────────┐
│                    │────▶│ Step 2   │ ✅
│                    │     └──────────┘
│                    │     ┌──────────┐
│                    │────▶│ Step 3   │ ❌ Failed!
│                    │     └──────────┘
└────────┬───────────┘
         │
         ▼
┌──────────────────┐
│    RE-PLANNER     │
│   Adjusts plan    │
│   3b. Alternative │
│   4. Synthesize   │
└──────────────────┘

class PlanAndExecuteAgent:
    def __init__(self, planner_llm, executor_llm, tools):
        self.planner = planner_llm   # Powerful model (e.g., GPT-4, Claude Opus)
        self.executor = executor_llm  # Fast model (e.g., GPT-4o-mini, Haiku)
        self.tools = tools

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

        # Phase 2: Execution
        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-plan if needed
                plan = self.replan(task, plan, results, i)

        # Phase 3: Synthesis
        return self.synthesize(task, results)

    def create_plan(self, task: str) -> list:
        prompt = f"""Create a step-by-step plan to accomplish this task.
Return a JSON list of steps.

Task: {task}

Available tools: {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"Step {i+1}: {r['result']}"
            for i, r in enumerate(previous_results)
        )
        prompt = f"""Execute this step using the available tools.

Context from previous steps:
{context}

Step to execute: {step['description']}"""

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

    def replan(self, task, original_plan, results, failed_index):
        prompt = f"""The original plan failed at step {failed_index + 1}.

Original plan: {json.dumps(original_plan)}
Results: {json.dumps(results)}

Create a new plan starting from step {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"""Synthesize the results to answer the task.
Task: {task}
Results: {json.dumps(results)}"""
        return self.planner.complete(prompt)

User task
         │
         ▼
    ┌─────────┐
    │  ACTOR   │ ← Produces a first response/solution
    └────┬────┘
         │
         ▼
    ┌─────────┐
    │EVALUATOR │ ← Analyzes quality (tests, criteria, score)
    └────┬────┘
         │
    ┌────▼────┐
    │ Score OK │──── YES ───▶ Final response ✅
    │    ?     │
    └────┬────┘
         │ NO
         ▼
    ┌──────────┐
    │REFLEXION  │ ← "The answer is missing X, the error is Y"
    └────┬─────┘
         │
         ▼
    ┌─────────┐
    │  ACTOR   │ ← New attempt with reflection as context
    └────┬────┘
         │
         ▼
       (loop)

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: Produce a response
            response = self.act(task, reflections)

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

            if evaluation["passed"]:
                return response

            # Phase 3: Reflect
            reflection = self.reflect(task, response, evaluation)
            reflections.append(reflection)

        # Return the best attempt
        return response

    def act(self, task: str, reflections: list) -> str:
        reflection_context = ""
        if reflections:
            reflection_context = "\n\nReflections on your previous attempts:\n"
            for i, r in enumerate(reflections):
                reflection_context += f"\nAttempt {i+1}: {r}\n"

        prompt = f"""Complete this task to the best of your ability.
{reflection_context}
Task: {task}"""
        return self.llm.complete(prompt)

    def evaluate(self, task: str, response: str) -> dict:
        # Can be an LLM, automated tests, or a score
        prompt = f"""Evaluate this response on a scale of 1-10.
Identify specific issues.

Task: {task}
Response: {response}

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

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

    def reflect(self, task, response, evaluation) -> str:
        prompt = f"""Analyze why this response is unsatisfactory
and provide specific instructions for improvement.

Task: {task}
Response: {response}
Issues: {evaluation['issues']}

Reflection:"""
        return self.llm.complete(prompt)

                    User task
                         │
                         ▼
                ┌─────────────────┐
                │   ORCHESTRATOR   │
                │   (Coordinator)  │
                └────────┬────────┘
                         │
          ┌──────────────┼──────────────┐
          ▼              ▼              ▼
   ┌────────────┐ ┌────────────┐ ┌────────────┐
   │  AGENT      │ │  AGENT      │ │  AGENT      │
   │  Research   │ │  Analysis   │ │  Writing    │
   │             │ │             │ │             │
   │  Tools:     │ │  Tools:     │ │  Tools:     │
   │  - Web      │ │  - Python   │ │  - Markdown │
   │  - News API │ │  - Charts   │ │  - CMS      │
   └──────┬─────┘ └──────┬─────┘ └──────┬─────┘
          │              │              │
          └──────────────┼──────────────┘
                         │
                         ▼
                  Final result

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"""You are {self.name}, a specialized agent.
Your role: {self.role}

Context from previous steps:
{context}

Task: {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:
        """Sequential execution: each agent passes its result to the next."""
        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:
        """The manager decides which agent to call and when."""
        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 delegations
            prompt = f"""You are the manager. Here's your team:
{worker_descriptions}

Overall task: {task}
Work done: {context}

Delegate the next subtask to an agent, or say DONE if everything is complete.
Format: DELEGATE:agent_name:subtask OR DONE:summary"""

            decision = manager.llm.complete(prompt)

            if decision.startswith("DONE"):
                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


# Usage
orchestrator = MultiAgentOrchestrator()

orchestrator.add_agent(Agent(
    "researcher", "Web research expert",
    llm=fast_llm, tools=[web_search, web_fetch]
))
orchestrator.add_agent(Agent(
    "analyst", "Data analysis expert",
    llm=smart_llm, tools=[python_exec, chart_gen]
))
orchestrator.add_agent(Agent(
    "writer", "Professional writer",
    llm=smart_llm, tools=[markdown_gen]
))
orchestrator.add_agent(Agent(
    "manager", "Project manager who coordinates the team",
    llm=smart_llm
))

# Hierarchical execution
result = orchestrator.run_hierarchical(
    "Write a report on AI trends in 2025",
    manager_name="manager"
)

User question
         │
         ▼
    ┌─────────────┐
    │  RAG AGENT   │
    │  (LLM)       │
    └──────┬──────┘
           │
           │ Dynamic source selection
           │
     ┌─────┼──────┬────────────┐
     ▼     ▼      ▼            ▼
┌────────┐┌────┐┌──────┐┌──────────┐
│Vector  ││SQL ││ Ext. ││  Web     │
│Search  ││    ││ API  ││ Search   │
│        ││    ││      ││          │
│Pinecone││PG  ││REST  ││Brave/    │
│Chroma  ││    ││      ││Google    │
└───┬────┘└──┬─┘└──┬───┘└────┬────┘
    │        │     │         │
    └────────┴─────┴─────────┘
                   │
                   ▼
           ┌──────────────┐
           │  Synthesis    │
           │  with 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:
        """Semantic search in a vector database."""
        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:
        """Execute a read-only SQL query."""
        # Validation: read-only!
        if not query.strip().upper().startswith("SELECT"):
            raise ValueError("Only SELECT queries are allowed")
        return self.db.execute(query).fetchall()

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

    def api_fetch(self, url: str, params: dict = None) -> dict:
        """Call an external API."""
        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"""You are a research agent.
Question: {question}

Sources found so far:
{json.dumps(sources, indent=2) if sources else "None"}

Available tools:
- vector_search(query, collection, top_k): Semantic search
- sql_query(query): SQL query
- web_search(query, num_results): Web search
- api_fetch(url, params): API call

Do you have enough information to answer? If yes, respond FINAL:your_answer
Otherwise, call a tool: TOOL:tool_name: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
                })

        # Forced synthesis after max_searches
        return self.synthesize(question, sources)

Is the task simple (1-3 steps)?
├── YES → ReAct
└── NO
    ├── Does quality need to be maximized?
    │   ├── YES → Reflexion
    │   └── NO
    │       ├── Does the task have predictable steps?
    │       │   ├── YES → Plan-and-Execute
    │       │   └── NO
    │       │       ├── Are varied skills needed?
    │       │       │   ├── YES → Multi-Agent
    │       │       │   └── NO → ReAct
    │       └── Is the task centered on information retrieval?
    │           └── YES → Tool-Augmented RAG