En tant qu'ingénieur qui a déployé des agents IA dans une douzaine de systèmes de production, je peux vous assurer que la mécanisme de réflexion personnelle — le "Reflexion" — est la différence entre un agent qui échoue silencieusement et un agent qui apprend de ses erreurs. Après avoir implémenté ce pattern sur HolySheep AI pour des clients处理 des milliers de requêtes quotidiennes, j'ai accumulé une expertise précieuse que je vais partager avec vous dans ce tutoriel approfondi.
Comprendre le Mécanisme de Reflexion
Le pattern Reflexion, introduit par Shinn et al. (2023), permet à un agent d'examiner ses propres actions passées, d'identifier les erreurs, et de formuler des stratégies correctives. Concrètement, cela signifie que votre agent ne se contente pas de générer une réponse — il évalue la qualité de cette réponse et l'améliore itérativement.
Architecture du Système
Voici l'architecture que j'ai fait évoluer au fil de mes déploiements :
┌─────────────────────────────────────────────────────────────────┐
│ ARCHITECTURE REFLEXION │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ User │───▶│ Agent │───▶│ Executor │ │
│ │ Query │ │ (Planner) │ │ (Tool Use) │ │
│ └──────────┘ └──────────────┘ └──────────────────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────────┐ │
│ │ Reflexor │◀────────│ Result │ │
│ │ (Self-Eval) │ │ Evaluation │ │
│ └──────────────┘ └──────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ Memory │ │
│ │ (Traject.) │ │
│ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘Cette architecture permet un cycle d'amélioration continue où chaque exécution est analysée et stockée pour les futures itérations.
Implémentation Complète du Moteur Reflexion
Passons maintenant au code. J'utilise HolySheep AI pour ce projet car leur infrastructure offre une latence moyenne de 47ms sur les appels de réflexion — essentielle pour maintenir la réactivité de l'agent.
1. Configuration de l'Environnement
# Installation des dépendances
pip install httpx asyncio pydantic aiofiles
Configuration des variables d'environnement
export HOLYSHEEP_API_KEY="YOUR_HOLYSHEEP_API_KEY"
export HOLYSHEEP_BASE_URL="https://api.holysheep.ai/v1"
Structure du projet
project/
├── reflexion/
│ ├── __init__.py
│ ├── agent.py # Agent principal avec Reflexion
│ ├── memory.py # Gestionnaire de mémoire
│ ├── evaluator.py # Module d'auto-évaluation
│ └── tools.py # Outils d'exécution
├── config.py
├── requirements.txt
└── main.py2. Module Principal de l'Agent avec Reflexion
# reflexion/agent.py
import asyncio
import json
from typing import List, Dict, Optional, Any
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import httpx
class ReflexionState(Enum):
PLANNING = "planning"
EXECUTING = "executing"
EVALUATING = "evaluating"
REFLECTING = "reflecting"
SUCCESS = "success"
MAX_ITERATIONS = "max_iterations"
@dataclass
class Action:
tool: str
input: Dict[str, Any]
timestamp: datetime = field(default_factory=datetime.now)
result: Optional[str] = None
@dataclass
class ReflexionContext:
objective: str
trajectory: List[Action] = field(default_factory=list)
reflections: List[str] = field(default_factory=list)
state: ReflexionState = ReflexionState.PLANNING
iteration: int = 0
max_iterations: int = 3
class ReflexionAgent:
"""
Agent avec mécanisme de自我反思 (Self-Reflexion).
Ce pattern permet à l'agent d'examiner ses actions passées,
d'identifier les erreurs, et de formuler des stratégies correctives.
"""
def __init__(
self,
api_key: str = "YOUR_HOLYSHEEP_API_KEY",
base_url: str = "https://api.holysheep.ai/v1",
model: str = "deepseek-v3.2",
max_iterations: int = 3,
temperature: float = 0.7
):
self.api_key = api_key
self.base_url = base_url
self.model = model
self.max_iterations = max_iterations
self.temperature = temperature
self.client = httpx.AsyncClient(timeout=120.0)
async def _call_llm(
self,
messages: List[Dict],
system_prompt: Optional[str] = None
) -> str:
"""Appel à l'API HolySheep avec optimisation des coûts."""
full_messages = []
if system_prompt:
full_messages.append({"role": "system", "content": system_prompt})
full_messages.extend(messages)
# Utilisation de DeepSeek V3.2 : $0.42/MTok (économie 85%+ vs GPT-4)
async with self.client.stream(
"POST",
f"{self.base_url}/chat/completions",
json={
"model": self.model,
"messages": full_messages,
"temperature": self.temperature,
"max_tokens": 2048
},
headers={"Authorization": f"Bearer {self.api_key}"}
) as response:
response.raise_for_status()
data = await response.json()
return data["choices"][0]["message"]["content"]
async def plan(self, context: ReflexionContext) -> str:
"""Génère un plan d'action basé sur l'objectif."""
prompt = f"""Tu es un agent de réflexion qui doit planifier des actions pour atteindre cet objectif :
Objectif : {context.objective}
Historique des actions précédentes :
{self._format_trajectory(context.trajectory)}
Réflexions précédentes :
{chr(10).join(context.reflections) if context.reflections else "Aucune"}
Analyse la situation et décide de la prochaine action à effectuer.
Réponds UNIQUEMENT avec un JSON valide :
{{"tool": "nom_de_l_outil", "input": {{"param": "valeur"}}, "reasoning": "explication"}}
"""
response = await self._call_llm([{"role": "user", "content": prompt}])
return self._parse_json_response(response)
async def execute(self, action: Dict, context: ReflexionContext) -> str:
"""Exécute l'action planifiée via les outils disponibles."""
tool_name = action.get("tool")
# Simulation d'exécution d'outil
# En production, remplacez par vos outils réels
result = f"Exécution de {tool_name} avec params: {action.get('input', {})}"
return result
async def reflect(self, context: ReflexionContext) -> str:
"""
Mécanisme de自我反思 : l'agent évalue ses actions passées.
Cette fonction analyse la trajectoire et génère une réflexion
critique pour améliorer les futures tentatives.
"""
prompt = f"""Tu es un agent en mode自我反思 (Self-Reflexion).
Analyse EXIGEAMMENT tes actions passées et identifie les erreurs potentielles.
Objectif original : {context.objective}
Actions exécutées :
{self._format_trajectory(context.trajectory)}
Pour chaque action, évalue :
1. L'efficacité : L'action a-t-elle fait progresser vers l'objectif ?
2. Les erreurs : Quels problèmes ont été commis ?
3. Les corrections : Comment éviter ces erreurs à l'avenir ?
Génère une réflexion structurée en JSON :
{{"evaluation": "globale de 1-10", "issues": ["liste des problèmes"], "improvements": ["corrections recommandées"], "confidence": "haute/moyenne/faible"}}
Sois HONNÊTE et critique dans ton évaluation. Ne te contente pas de valider."""
reflection = await self._call_llm([{"role": "user", "content": prompt}])
parsed = self._parse_json_response(reflection)
# Stocker la réflexion pour référence future
context.reflections.append(
f"Iteration {context.iteration}: {reflection[:200]}"
)
return parsed
async def should_continue(self, reflection: Dict, context: ReflexionContext) -> bool:
"""Détermine si l'agent doit continuer à itérer."""
confidence = reflection.get("confidence", "moyenne")
evaluation = int(reflection.get("evaluation", "5").replace("/10", ""))
# Critères de terminaison
if context.iteration >= self.max_iterations:
return False
if confidence == "haute" and evaluation >= 8:
return False
if evaluation >= 9:
return False
return True
async def run(self, objective: str) -> Dict[str, Any]:
"""Exécute le cycle complet Reflexion."""
context = ReflexionContext(objective=objective)
while context.iteration < self.max_iterations:
context.iteration += 1
context.state = ReflexionState.PLANNING
# Étape 1: Planification
plan = await self.plan(context)
context.state = ReflexionState.EXECUTING
# Étape 2: Exécution
result = await self.execute(plan, context)
context.trajectory.append(Action(
tool=plan.get("tool", "unknown"),
input=plan.get("input", {}),
result=result
))
# Étape 3: Auto-évaluation (Reflexion)
context.state = ReflexionState.REFLECTING
reflection = await self.reflect(context)
# Étape 4: Décision de continuation
if not await self.should_continue(reflection, context):
context.state = ReflexionState.SUCCESS
break
if context.iteration >= self.max_iterations:
context.state = ReflexionState.MAX_ITERATIONS
return self._format_result(context)
def _format_trajectory(self, trajectory: List[Action]) -> str:
if not trajectory:
return "Aucune action encore exécutée."
return "\n".join([
f"- {i+1}. {a.tool}: {a.input} -> {a.result[:100] if a.result else 'N/A'}..."
for i, a in enumerate(trajectory)
])
def _parse_json_response(self, response: str) -> Dict:
"""Parse la réponse JSON du LLM."""
try:
# Extraction du JSON depuis la réponse
start = response.find("{")
end = response.rfind("}") + 1
if start != -1 and end > start:
return json.loads(response[start:end])
except json.JSONDecodeError:
pass
return {"raw": response}
def _format_result(self, context: ReflexionContext) -> Dict[str, Any]:
return {
"objective": context.objective,
"final_state": context.state.value,
"iterations": context.iteration,
"trajectory": [
{"tool": a.tool, "input": a.input, "result": a.result}
for a in context.trajectory
],
"reflections": context.reflections
}
Exemple d'utilisation
async def main():
agent = ReflexionAgent(
api_key="YOUR_HOLYSHEEP_API_KEY",
model="deepseek-v3.2"
)
result = await agent.run(
"Rédiger un rapport sur les tendances IA 2026"
)
print(f"Résultat final: {result['final_state']}")
print(f"Itérations: {result['iterations']}")
for reflection in result['reflections']:
print(f" - {reflection}")
if __name__ == "__main__":
asyncio.run(main())3. Gestionnaire de Mémoire avec Persistance
# reflexion/memory.py
import asyncio
import json
import aiofiles
from pathlib import Path
from typing import List, Dict, Optional, Any
from datetime import datetime
from collections import defaultdict
import hashlib
class ReflexionMemory:
"""
Gestionnaire de mémoire pour stocker et récupérer
les trajectoires de réflexion. Implémente un système
de Vector Store simplifié pour la similarité.
"""
def __init__(self, storage_path: str = "./memory_store"):
self.storage_path = Path(storage_path)
self.storage_path.mkdir(parents=True, exist_ok=True)
self.index_file = self.storage_path / "index.json"
self.memories: Dict[str, List[Dict]] = defaultdict(list)
self._lock = asyncio.Lock()
async def initialize(self):
"""Charge l'index existant au démarrage."""
if self.index_file.exists():
async with aiofiles.open(self.index_file, 'r') as f:
content = await f.read()
self.memories = defaultdict(list, json.loads(content))
async def store(
self,
key: str,
experience: Dict[str, Any],
metadata: Optional[Dict] = None
):
"""Stocke une expérience de réflexion avec méta-données."""
async with self._lock:
entry = {
"timestamp": datetime.now().isoformat(),
"experience": experience,
"metadata": metadata or {},
"embedding_key": self._generate_key(experience)
}
self.memories[key].append(entry)
await self._save_index()
async def retrieve(
self,
key: str,
limit: int = 5,
similarity_threshold: float = 0.7
) -> List[Dict]:
"""Récupère les expériences les plus pertinentes."""
async with self._lock:
experiences = self.memories.get(key, [])
# Tri par timestamp (plus récent d'abord)
sorted_exp = sorted(
experiences,
key=lambda x: x["timestamp"],
reverse=True
)
return sorted_exp[:limit]
async def retrieve_similar(
self,
query: str,
limit: int = 5
) -> List[Dict]:
"""Récupère les expériences similaires à une requête."""
query_key = self._generate_key({"text": query})
similar = []
for key, experiences in self.memories.items():
for exp in experiences:
similarity = self._calculate_similarity(
query_key,
exp["embedding_key"]
)
if similarity >= 0.7:
similar.append({
**exp,
"similarity": similarity
})
return sorted(similar, key=lambda x: x["similarity"], reverse=True)[:limit]
async def get_statistics(self) -> Dict[str, Any]:
"""Retourne des statistiques sur la mémoire."""
total_experiences = sum(len(v) for v in self.memories.values())
return {
"total_keys": len(self.memories),
"total_experiences": total_experiences,
"average_per_key": total_experiences / max(len(self.memories), 1),
"storage_size_mb": sum(
f.stat().st_size
for f in self.storage_path.glob("*.json")
) / (1024 * 1024)
}
def _generate_key(self, data: Dict) -> str:
"""Génère une clé de similarité (simplifié pour la démo)."""
content = json.dumps(data, sort_keys=True)
return hashlib.md5(content.encode()).hexdigest()[:16]
def _calculate_similarity(self, key1: str, key2: str) -> float:
"""Calcule la similarité entre deux clés (Jaccard simplifié)."""
if not key1 or not key2:
return 0.0
set1, set2 = set(key1), set(key2)
intersection = len(set1 & set2)
union = len(set1 | set2)
return intersection / union if union > 0 else 0.0
async def _save_index(self):
"""Sauvegarde l'index de manière atomique."""
temp_file = self.storage_path / f"index_temp_{id(self)}.json"
async with aiofiles.open(temp_file, 'w') as f:
await f.write(json.dumps(dict(self.memories)))
temp_file.replace(self.index_file)
Intégration avec l'Agent
class ReflexionAgentWithMemory(ReflexionAgent):
"""Agent Reflexion avec persistance de mémoire."""
def __init__(self, *args, memory_path: str = "./memory", **kwargs):
super().__init__(*args, **kwargs)
self.memory = ReflexionMemory(memory_path)
async def run(self, objective: str) -> Dict[str, Any]:
"""Exécute avec apprentissage mémoire."""
# Récupération des expériences similaires
similar = await self.memory.retrieve_similar(objective)
result = await super().run(objective)
# Stockage de l'expérience pour apprentissage futur
await self.memory.store(
key=self._generate_context_key(objective),
experience=result,
metadata={
"objective": objective,
"success": result["final_state"] == "success"
}
)
return result
def _generate_context_key(self, objective: str) -> str:
"""Génère une clé de contexte pour le stockage."""
# Utilise les premiers mots comme clé de catégorie
words = objective.lower().split()[:3]
return "_".join(words)
Benchmark de performance mémoire
async def benchmark_memory():
"""Benchmark du système de mémoire."""
import time
memory = ReflexionMemory("./benchmark_memory")
await memory.initialize()
# Test d'écriture concurrente
num_operations = 1000
start = time.perf_counter()
tasks = [
memory.store(
f"task_{i % 100}",
{"result": f"experiment_{i}", "score": i / 10},
{"batch": "benchmark"}
)
for i in range(num_operations)
]
await asyncio.gather(*tasks)
write_time = time.perf_counter() - start
write_ops_per_sec = num_operations / write_time
# Test de lecture
start = time.perf_counter()
for _ in range(100):
await memory.retrieve("task_0", limit=10)
read_time = time.perf_counter() - start
read_ops_per_sec = 100 / read_time
print(f"=== Benchmark Mémoire ===")
print(f"Écritures: {write_ops_per_sec:.0f} ops/sec ({write_time:.3f}s pour {num_operations})")
print(f"Lectures: {read_ops_per_sec:.0f} ops/sec")
print(f"Statistiques: {await memory.get_statistics()}")
if __name__ == "__main__":
asyncio.run(benchmark_memory())Optimisation des Performances et Contrôle de Concurrence
Lors de mes déploiements en production, j'ai rencontrés des défis significatifs de performance. Voici les optimisations que j'ai développées :
Gestionnaire de Concurrence Multi-Agent
# reflexion/concurrency.py
import asyncio
from typing import List, Dict, Optional, Any
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import deque
import time
@dataclass
class RateLimiter:
"""Limiteur de taux avec fenêtre glissante."""
max_requests: int
window_seconds: int
_requests: deque = field(default_factory=deque)
async def acquire(self) -> float:
"""Acquiert un token, retourne le temps d'attente si limité."""
now = datetime.now()
cutoff = now - timedelta(seconds=self.window_seconds)
# Nettoyage des requêtes expirées
while self._requests and self._requests[0] < cutoff:
self._requests.popleft()
if len(self._requests) >= self.max_requests:
# Calcul