Introduction aux Agents CrewAI
CrewAI est un framework Python puissant qui permet de créer des systèmes multi-agents sophistiqués pour l'automatisation de tâches complexes. La personnalisation du comportement des agents repose principalement sur une conception minutieuse des prompts et une configuration appropriée des rôles.
Installation et Configuration
Installation de CrewAI
pip install crewai
Installation des dépendances optionnelles
pip install 'crewai[tools]'
Architecture des Agents Personnalisés
Définition d'un Agent avec Comportement Personnalisé
from crewai import Agent
Agent analyste avec comportement spécialisé
analyste = Agent(
role="Analyste de données marketing",
goal="Extraire des insights actionables des données clients",
backstory="""
Vous êtes un analyste senior avec 10 ans d'expérience en marketing digital.
Votre expertise inclut l'analyse de données, la détection de patterns et la
formulation de recommandations stratégiques basées sur les données.
""",
verbose=True,
allow_delegation=True,
tools=[
# Outils personnalisés pour l'agent
]
)
Techniques Avancées de Prompt Engineering
1. Context Injection Dynamique
def generer_prompt_contextuel(agent, contexte_supplementaire):
"""Génère un prompt enrichi avec du contexte dynamique"""
prompt_base = agent.backstory
# Injection de contexte spécifique au domaine
contexte_domaine = """
Contexte actuel du marché:
- Secteur: E-commerce
- Période d'analyse: Q1 2026
- Objectif: Optimiser les conversions
Contraintes:
- Budget limité à 50 000€
- Délai: 2 semaines
"""
return f"{prompt_base}\n\n{contexte_domaine}\n\n{contexte_supplementaire}"
Utilisation avec un agent
agent_contextualise = Agent(
role="Stratégiste marketing",
goal="Optimiser les performances des campagnes",
backstory=generer_prompt_contextuel(
agent_base,
"Focus sur les canaux numériques à fort ROI"
)
)
2. System Prompts Structurés
SYSTEM_PROMPT_TEMPLATE = """
RÔLE
Tu es {role}
OBJECTIF PRINCIPAL
{goal}
CONTRAINTES
- {constraints}
- Ne jamais divulguer d'informations sensibles
- Respecter les délais impartis
FORMAT DE RÉPONSE
{response_format}
EXEMPLES
{examples}
"""
def creer_agent_structures(role, goal, constraints, format_reponse, examples):
"""Crée un agent avec un prompt structuré et cohérent"""
prompt = SYSTEM_PROMPT_TEMPLATE.format(
role=role,
goal=goal,
constraints="\n- ".join(constraints) if isinstance(constraints, list) else constraints,
response_format=format_reponse,
examples="\n".join([f"- {ex}" for ex in examples])
)
return Agent(
role=role,
goal=goal,
backstory=prompt,
verbose=True
)
Patterns de Comportement Multi-Agents
Orchestration Hiérarchique
from crewai import Crew, Task, Process
Définition des agents spécialisés
chef_projet = Agent(
role="Chef de projet IA",
goal="Coordonner l'équipe et valider les livrables",
backstory="Expert en gestion de projets IA avec une vision stratégique"
)
developpeur = Agent(
role="Développeur Python",
goal="Implémenter des solutions techniques robustes",
backstory="Développeur senior Python, expert en frameworks ML/IA"
)
analyste = Agent(
role="Analyste de données",
goal="Extraire et analyser les données pertinentes",
backstory="Data scientist avec expertise en analyse statistique"
)
Configuration du crew avec processus hiérarchique
crew = Crew(
agents=[chef_projet, developpeur, analyste],
tasks=[
# Tâches ordonnées avec dépendances
],
process=Process.hierarchical,
manager_agent=chef_projet
)
Exécution
resultat = crew.kickoff()
Gestion des États et Mémoire
from crewai.memory import Memory, ShortTermMemory, LongTermMemory
class AgentAvecMemoire:
"""Classe wrapper pour ajouter de la mémoire persistante"""
def __init__(self, agent, type_memoire='long_term'):
self.agent = agent
self.memory = Memory(
type=type_memoire,
storage_backend="redis" # Configuration Redis optionnelle
)
def executer_avec_memoire(self, entree):
"""Exécute l'agent avec contexte de la mémoire"""
# Récupération du contexte pertinent
contexte = self.memory.get_relevant_context(entree)
# Enrichissement de la requête
requete_enrichie = f"""
Contexte mémoire:
{contexte}
Nouvelle requête:
{entree}
"""
# Exécution avec contexte
resultat = self.agent.execute(requete_enrichie)
# Sauvegarde en mémoire
self.memory.add(
input=entree,
output=resultat,
metadata={'timestamp': datetime.now().isoformat()}
)
return resultat
Optimisation des Performances
Configuration des Paramètres de Génération
from crewai.llm import LLMConfig
Configuration optimisée pour différents cas d'usage
CONFIGURATIONS_LLM = {
'precis': {
'temperature': 0.1,
'top_p': 0.9,
'frequency_penalty': 0.5,
'presence_penalty': 0.3
},
'creatif': {
'temperature': 0.8,
'top_p': 0.95,
'frequency_penalty': 0.2,
'presence_penalty': 0.4
},
'equilibre': {
'temperature': 0.5,
'top_p': 0.92,
'frequency_penalty': 0.3,
'presence_penalty': 0.2
}
}
def creer_agent_optimise(role, goal, config_name='equilibre'):
"""Crée un agent avec configuration LLM optimisée"""
config = CONFIGURATIONS_LLM.get(config_name, CONFIGURATIONS_LLM['equilibre'])
return Agent(
role=role,
goal=goal,
backstory="",
llm_config=LLMConfig(
model="gpt-4",
**config
)
)
Débogage et Monitoring
import logging
from crewai.utilities import Printer
Configuration du logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('crewai.custom')
class DebugAgent(Agent):
"""Agent avec capacités de débogage intégrées"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.printer = Printer()
self.execution_count = 0
def execute(self, task):
self.execution_count += 1
logger.info(f"Exécution #{self.execution_count} - Agent: {self.role}")
try:
result = super().execute(task)
logger.info(f"Succès - Durée: {result.execution_time}s")
return result
except Exception as e:
logger.error(f"Erreur: {str(e)}")
self._analyser_erreur(task, e)
raise
def _analyser_erreur(self, task, erreur):
"""Analyse les erreurs pour améliorer les prompts"""
suggestions = {
'context_length': "Augmenter le contexte ou diviser la tâche",
'timeout': "Réduire la complexité ou augmenter le timeout",
'rate_limit': "Implémenter un backoff exponentiel"
}
type_erreur = self._detecter_type_erreur(erreur)
logger.warning(f"Suggestion: {suggestions.get(type_erreur, 'Vérifier les paramètres')}")
Erreurs Courantes et Solutions
Erreur 1 : Contexte de Fenêtre Dépassé
Erreur: Token limit exceeded - maximum context window reached
**Solution :**
Implémentation du chunking intelligent
def chunker_texte(texte, limite_tokens=4000, chevauchement=200):
"""Découpe le texte en chunks avec chevauchement"""
encoding = tiktoken.get_encoding("cl100k_base")
tokens = encoding.encode(texte)
chunks = []
for i in range(0, len(tokens), limite_tokens - chevauchement):
chunk_tokens = tokens[i:i + limite_tokens]
chunk_texte = encoding.decode(chunk_tokens)
chunks.append(chunk_texte)
return chunks
Utilisation avec un agent
def traiter_texte_long(agent, texte):
chunks = chunker_texte(texte)
resultats = []
for i, chunk in enumerate(chunks):
logger.info(f"Traitement du chunk {i+1}/{len(chunks)}")
resultat = agent.execute(Task(description=chunk))
resultats.append(resultat)
return fusionner_resultats(resultats)
Erreur 2 : Agents Bloqués en Boucle
Erreur: Agent execution timeout - circular dependency detected
**Solution :**
from functools import lru_cache
class AgentAvecAntiBoucle:
"""Agent avec détection de boucles"""
def __init__(self, agent, max_iterations=5):
self.agent = agent
self.max_iterations = max_iterations
self.execution_history = []
def execute_securise(self, task):
"""Exécution avec détection de boucles"""
iteration = 0
resultat_precedent = None
while iteration < self.max_iterations:
resultat = self.agent.execute(task)
# Vérification de la progression
if self._est_boucle(resultat, resultat_precedent):
logger.warning("Boucle détectée - modification du prompt")
task = self._modifier_pour_sortir_boucle(task)
else:
return resultat
resultat_precedent = resultat
iteration += 1
raise RuntimeError("Limite d'itérations atteinte")
def _est_boucle(self, resultat_courant, resultat_precedent):
"""Détecte si deux résultats sont trop similaires"""
if resultat_precedent is None:
return False
similarite = self._calculer_similarite(
resultat_courant.text,
resultat_precedent.text
)
return similarite > 0.95
def _calculer_similarite(self, texte1, texte2):
"""Calcule la similarité entre deux textes"""
from difflib import SequenceMatcher
return SequenceMatcher(None, texte1, texte2).ratio()
Erreur 3 : Rate Limiting API
Erreur: 429 Too Many Requests - rate limit exceeded
**Solution :**
import time
from functools import wraps
def retry_with_backoff(max_retries=5, initial_delay=1):
"""Décorateur pour gérer les rate limits avec backoff exponentiel"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
delay = initial_delay
for tentative in range(max_retries):
try:
return func(*args, **kwargs)
except RateLimitError as e:
if tentative == max_retries - 1:
raise
logger.warning(f"Rate limit - attente {delay}s")
time.sleep(delay)
delay *= 2 # Backoff exponentiel
return None
return wrapper
return decorator
class AgentAvecRateLimit:
"""Agent avec gestion intelligente des rate limits"""
def __init__(self, agent, requests_per_minute=60):
self.agent = agent
self.rate_limiter = TokenBucket(capacity=requests_per_minute)
@retry_with_backoff(max_retries=5)
def execute(self, task):
"""Exécution avec limitation de débit"""
with self.rate_limiter:
return self.agent.execute(task)
Meilleures Pratiques
Organisation du Code
"""
Structure recommandée pour un projet CrewAI
"""
projet/
├── agents/
│ ├── __init__.py
│ ├── base_agent.py
│ ├── analyste.py
│ └── developpeur.py
├── crews/
│ ├── __init__.py
│ ├── crew_principal.py
│ └── crew_secondary.py
├── tasks/
│ ├── __init__.py
│ └── task_definitions.py
├── prompts/
│ ├── system_prompts.py
│ └── templates.py
├── utils/
│ ├── __init__.py
│ ├── memory.py
│ └── monitoring.py
└── config/
├── settings.py
└── llm_config.yaml
Conclusion
La personnalisation des agents CrewAI nécessite une approche méthodique combinant une conception rigoureuse des prompts, une gestion sophistiquée de la mémoire et des mécanismes robustes de gestion des erreurs. En appliquant ces techniques, vous pouvez créer des systèmes multi-agents capables de résoudre des tâches complexes de manière autonome et efficace.
L'expérimentation est clé : testez différentes configurations, mesurez les performances et itérez continuellement pour optimiser vos agents selon vos cas d'usage spécifiques.
---
*Article publié sur HolySheep AI Blog - Tous droits réservés*
Ressources connexes
Articles connexes