LLM et IA Générative
Agents IA Avancés
Architectures avancées d'agents - MCP, orchestration multi-agents, ReAct, planification et systèmes autonomes
Agents IA Avancés
Au-delà du function calling basique, les agents modernes utilisent des architectures sophistiquées pour résoudre des tâches complexes de manière autonome.
Model Context Protocol (MCP)
Le MCP (Anthropic) est un protocole standardisé pour connecter les LLM à des sources de données et outils externes.
┌─────────────────────────────────────────────────────────────────┐
│ MODEL CONTEXT PROTOCOL │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ MCP Protocol ┌─────────────────────┐ │
│ │ │ ◄──────────────────► │ MCP Server │ │
│ │ Claude │ │ (Python/Node) │ │
│ │ │ JSON-RPC ├─────────────────────┤ │
│ └─────────────┘ │ • Resources │ │
│ │ • Tools │ │
│ │ • Prompts │ │
│ └─────────────────────┘ │
│ │ │
│ ┌────────────────────┼────────────┐│
│ │ │ ││
│ ▼ ▼ ▼│
│ ┌────────┐ ┌────────┐ ┌────────┐
│ │Database│ │ API │ │ Files │
│ └────────┘ └────────┘ └────────┘
│ │
└─────────────────────────────────────────────────────────────────┘Architecture MCP
# server.py - Serveur MCP
from mcp.server import Server
from mcp.types import Tool, Resource
server = Server("my-mcp-server")
# Définir un outil
@server.tool("search_database")
async def search_database(query: str, limit: int = 10):
"""Recherche dans la base de données"""
results = await db.search(query, limit=limit)
return {"results": results}
# Définir une ressource
@server.resource("user://{user_id}")
async def get_user(user_id: str):
"""Récupère les informations d'un utilisateur"""
user = await db.get_user(user_id)
return Resource(
uri=f"user://{user_id}",
content=user.to_json(),
mime_type="application/json"
)
# Démarrer le serveur
if __name__ == "__main__":
server.run()Configuration client
{
"mcpServers": {
"database": {
"command": "python",
"args": ["server.py"],
"env": {
"DATABASE_URL": "postgresql://..."
}
},
"filesystem": {
"command": "npx",
"args": ["@anthropic/mcp-server-filesystem", "/path/to/files"]
}
}
}Avantages de MCP
| Feature | Description |
|---|---|
| Standardisé | Un protocole pour tous les outils |
| Bidirectionnel | Le serveur peut notifier le client |
| Sécurisé | Isolation et permissions granulaires |
| Réutilisable | Serveurs partagés entre applications |
Patterns d'agents avancés
ReAct (Reasoning + Acting)
Alterne entre raisonnement et action.
┌─────────────────────────────────────────────────────────────────┐
│ REACT PATTERN │
├─────────────────────────────────────────────────────────────────┤
│ │
│ Question: "Quel est le PIB par habitant de la France ?" │
│ │
│ Thought 1: Je dois trouver le PIB et la population │
│ Action 1: search("PIB France 2024") │
│ Observation 1: PIB France = 2,780 milliards € │
│ │
│ Thought 2: Maintenant je cherche la population │
│ Action 2: search("population France 2024") │
│ Observation 2: Population = 68 millions │
│ │
│ Thought 3: Je peux calculer le PIB/habitant │
│ Action 3: calculate("2780000000000 / 68000000") │
│ Observation 3: 40,882 │
│ │
│ Final: Le PIB par habitant est d'environ 40,882 € │
│ │
└─────────────────────────────────────────────────────────────────┘from langchain.agents import initialize_agent, Tool
from langchain.agents import AgentType
tools = [
Tool(name="Search", func=search, description="Recherche web"),
Tool(name="Calculate", func=calculate, description="Calculs"),
]
agent = initialize_agent(
tools,
llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True,
)
result = agent.run("Quel est le PIB par habitant de la France ?")Plan-and-Execute
Planifie d'abord, exécute ensuite.
┌─────────────────────────────────────────────────────────────────┐
│ PLAN-AND-EXECUTE │
├─────────────────────────────────────────────────────────────────┤
│ │
│ PHASE 1: PLANIFICATION │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Tâche: "Créer un rapport de ventes Q4" │ │
│ │ │ │
│ │ Plan: │ │
│ │ 1. Récupérer données ventes Q4 depuis DB │ │
│ │ 2. Calculer métriques (CA, croissance, top produits) │ │
│ │ 3. Générer graphiques │ │
│ │ 4. Rédiger analyse │ │
│ │ 5. Compiler en PDF │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ PHASE 2: EXÉCUTION (séquentielle ou parallèle) │
│ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ │
│ │ 1 │→ │ 2 │→ │ 3 │→ │ 4 │→ │ 5 │ │
│ └─────┘ └─────┘ └─────┘ └─────┘ └─────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘from langchain_experimental.plan_and_execute import (
PlanAndExecute,
load_agent_executor,
load_chat_planner,
)
planner = load_chat_planner(llm)
executor = load_agent_executor(llm, tools, verbose=True)
agent = PlanAndExecute(planner=planner, executor=executor)
result = agent.run("Créer un rapport de ventes Q4")Reflexion
L'agent apprend de ses erreurs.
class ReflexionAgent:
def __init__(self, llm, tools, memory):
self.llm = llm
self.tools = tools
self.memory = memory # Stocke les tentatives passées
async def run(self, task, max_attempts=3):
for attempt in range(max_attempts):
# Exécuter la tâche
result = await self.execute(task)
# Évaluer le résultat
evaluation = await self.evaluate(task, result)
if evaluation.success:
return result
# Réfléchir sur l'échec
reflection = await self.reflect(task, result, evaluation)
# Stocker pour la prochaine tentative
self.memory.add(reflection)
return self.memory.best_attempt()
async def reflect(self, task, result, evaluation):
prompt = f"""
Tâche: {task}
Résultat: {result}
Problème: {evaluation.error}
Analyse ce qui n'a pas fonctionné et propose une meilleure approche.
"""
return await self.llm.generate(prompt)Systèmes multi-agents
Architecture hiérarchique
┌─────────────────────────────────────────────────────────────────┐
│ MULTI-AGENT HIERARCHY │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────┐ │
│ │ Orchestrator │ │
│ │ (Manager) │ │
│ └────────┬────────┘ │
│ │ │
│ ┌─────────────────┼─────────────────┐ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ Research │ │ Writer │ │ Reviewer │ │
│ │ Agent │ │ Agent │ │ Agent │ │
│ └────────────┘ └────────────┘ └────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘from crewai import Agent, Task, Crew
# Définir les agents
researcher = Agent(
role="Chercheur",
goal="Trouver des informations pertinentes",
backstory="Expert en recherche documentaire",
tools=[search_tool, web_scraper],
)
writer = Agent(
role="Rédacteur",
goal="Rédiger du contenu clair et engageant",
backstory="Journaliste expérimenté",
)
reviewer = Agent(
role="Relecteur",
goal="Vérifier la qualité et l'exactitude",
backstory="Éditeur senior avec 20 ans d'expérience",
)
# Définir les tâches
research_task = Task(
description="Rechercher les dernières tendances en IA",
agent=researcher,
)
write_task = Task(
description="Rédiger un article basé sur la recherche",
agent=writer,
context=[research_task], # Dépend de la recherche
)
review_task = Task(
description="Relire et améliorer l'article",
agent=reviewer,
context=[write_task],
)
# Créer l'équipe
crew = Crew(
agents=[researcher, writer, reviewer],
tasks=[research_task, write_task, review_task],
verbose=True,
)
result = crew.kickoff()Communication inter-agents
from autogen import AssistantAgent, UserProxyAgent, GroupChat
# Agents spécialisés
coder = AssistantAgent(
name="Coder",
system_message="Tu es un développeur Python expert.",
)
critic = AssistantAgent(
name="Critic",
system_message="Tu analyses le code et suggères des améliorations.",
)
tester = AssistantAgent(
name="Tester",
system_message="Tu écris des tests unitaires.",
)
# Chat de groupe
groupchat = GroupChat(
agents=[coder, critic, tester],
messages=[],
max_round=10,
)
# Le manager orchestre la conversation
manager = autogen.GroupChatManager(groupchat=groupchat)
# Lancer la discussion
coder.initiate_chat(
manager,
message="Implémente une fonction de tri rapide",
)Outils et orchestration
LangGraph
Graphes d'états pour agents complexes.
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
class AgentState(TypedDict):
messages: list
next_step: str
def should_continue(state: AgentState) -> str:
"""Décide si on continue ou termine"""
if "FINAL" in state["messages"][-1]:
return "end"
return "continue"
# Créer le graphe
workflow = StateGraph(AgentState)
# Ajouter les nœuds
workflow.add_node("research", research_node)
workflow.add_node("analyze", analyze_node)
workflow.add_node("synthesize", synthesize_node)
# Définir les transitions
workflow.add_edge("research", "analyze")
workflow.add_conditional_edges(
"analyze",
should_continue,
{
"continue": "research",
"end": "synthesize",
}
)
workflow.add_edge("synthesize", END)
# Compiler et exécuter
app = workflow.compile()
result = app.invoke({"messages": ["Analyse le marché de l'IA"]})Claude Computer Use
Contrôle d'ordinateur par l'IA.
import anthropic
client = anthropic.Anthropic()
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
tools=[
{
"type": "computer_20241022",
"name": "computer",
"display_width_px": 1024,
"display_height_px": 768,
},
{
"type": "text_editor_20241022",
"name": "str_replace_editor",
},
{
"type": "bash_20241022",
"name": "bash",
},
],
messages=[
{
"role": "user",
"content": "Ouvre Firefox et recherche 'météo Paris'"
}
],
)Mémoire et contexte
Types de mémoire
┌─────────────────────────────────────────────────────────────────┐
│ TYPES DE MÉMOIRE │
├─────────────────────────────────────────────────────────────────┤
│ │
│ SHORT-TERM LONG-TERM EPISODIC │
│ ┌───────────────┐ ┌───────────────┐ ┌───────────┐│
│ │ Contexte │ │ Vector DB │ │ Sessions ││
│ │ conversation │ │ (embeddings) │ │ passées ││
│ │ │ │ │ │ ││
│ │ • Derniers │ │ • Faits │ │ • Actions ││
│ │ messages │ │ • Documents │ │ • Résults ││
│ │ • État actuel │ │ • Procédures │ │ • Erreurs ││
│ └───────────────┘ └───────────────┘ └───────────┘│
│ │
└─────────────────────────────────────────────────────────────────┘from langchain.memory import (
ConversationBufferMemory,
VectorStoreRetrieverMemory,
CombinedMemory,
)
# Mémoire court-terme
short_term = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True,
)
# Mémoire long-terme (vector store)
long_term = VectorStoreRetrieverMemory(
retriever=vectorstore.as_retriever(),
memory_key="long_term_memory",
)
# Combiner les deux
memory = CombinedMemory(memories=[short_term, long_term])Sécurité des agents
Sandboxing
import docker
def execute_code_safely(code: str) -> str:
"""Exécute du code dans un container isolé"""
client = docker.from_env()
container = client.containers.run(
"python:3.11-slim",
command=f'python -c "{code}"',
mem_limit="512m",
cpu_period=100000,
cpu_quota=50000, # 50% CPU max
network_disabled=True, # Pas de réseau
remove=True,
timeout=30,
)
return container.decode("utf-8")Validation des actions
class ActionValidator:
def __init__(self, allowed_actions: list, blocked_patterns: list):
self.allowed = allowed_actions
self.blocked = blocked_patterns
def validate(self, action: dict) -> bool:
# Vérifier si l'action est autorisée
if action["type"] not in self.allowed:
raise SecurityError(f"Action non autorisée: {action['type']}")
# Vérifier les patterns bloqués
for pattern in self.blocked:
if pattern in str(action):
raise SecurityError(f"Pattern bloqué détecté: {pattern}")
return True
validator = ActionValidator(
allowed_actions=["search", "read_file", "write_file"],
blocked_patterns=["rm -rf", "sudo", "DROP TABLE"],
)Pour aller plus loin
- MCP Specification - Anthropic
- LangGraph - State machines
- CrewAI - Multi-agent framework
- AutoGen - Microsoft
- ReAct Paper - Yao et al.