CrewAI : Equipes d'agents RAG specialises

CrewAI est un framework Python pour orchestrer des equipes d'agents IA autonomes. Chaque agent a un role specifique avec des competences et des outils dedies. Ensemble, ils collaborent pour accomplir des taches complexes comme un RAG avance.

Prerequis : Consultez notre guide sur l'orchestration d'agents RAG pour comprendre les fondamentaux.

Pourquoi CrewAI pour le RAG ?

L'approche "Equipe"

CrewAI modelise des equipes comme dans le monde reel : chaque membre a un role, des competences et des outils specifiques.

ARCHITECTURE CREWAI

┌────────────────────────────────────────────────────┐
│                      CREW                           │
├────────────────────────────────────────────────────┤
│                                                    │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────┐  │
│  │   AGENT 1    │  │   AGENT 2    │  │  AGENT 3 │  │
│  │  Researcher  │  │   Analyst    │  │  Writer  │  │
│  ├──────────────┤  ├──────────────┤  ├──────────┤  │
│  │ Role: Search │  │ Role: Verify │  │Role:Write│  │
│  │ Goal: Find   │  │ Goal: Check  │  │Goal:Draft│  │
│  │ Tools: RAG   │  │ Tools: None  │  │Tools:None│  │
│  └──────┬───────┘  └──────┬───────┘  └────┬─────┘  │
│         │                 │               │        │
│         ▼                 ▼               ▼        │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────┐  │
│  │   TASK 1     │  │   TASK 2     │  │  TASK 3  │  │
│  │  Research    │  │   Analysis   │  │  Writing │  │
│  │  documents   │  │   & verify   │  │  response│  │
│  └──────────────┘  └──────────────┘  └──────────┘  │
│                                                    │
└────────────────────────────────────────────────────┘

Comparaison avec les autres frameworks

Avantages de CrewAI

• Simplicite : API intuitive basee sur des concepts familiers (roles, taches)
• Modularite : Agents reutilisables entre differentes crews
• Delegation : Agents peuvent deleguer automatiquement entre eux
• Outils : Integration facile d'outils personnalises

Concepts fondamentaux

Agents, Tasks et Crew

DEVELOPERpython
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

# Configuration LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

# 1. Definition des Agents
researcher = Agent(
    role="Chercheur Documentaire Senior",
    goal="Trouver des informations precises et pertinentes sur le sujet demande",
    backstory="""Tu es un chercheur experimente avec 10 ans d'experience
    dans l'analyse de documents techniques. Tu excelles a extraire
    les informations cles et a identifier les sources fiables.
    Tu ne te contentes pas des premiers resultats et approfondis
    toujours tes recherches.""",
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

analyst = Agent(
    role="Analyste de Contenu",
    goal="Synthetiser et verifier les informations trouvees",
    backstory="""Tu es un analyste critique avec une expertise
    en verification des faits. Tu identifies les contradictions,
    evalues la fiabilite des sources et structures les informations
    de maniere logique.""",
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

writer = Agent(
    role="Redacteur Technique Expert",
    goal="Produire des reponses claires, structurees et bien sourcees",
    backstory="""Tu es un redacteur technique avec une expertise
    en vulgarisation. Tu transformes des informations complexes
    en contenus accessibles et engageants tout en maintenant
    la precision technique.""",
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

# 2. Definition des Tasks
research_task = Task(
    description="""Recherche des informations sur: {topic}

    Tu dois:
    1. Identifier les concepts cles lies au sujet
    2. Rechercher dans la base de connaissances
    3. Collecter 3-5 sources pertinentes
    4. Extraire les informations principales de chaque source
    5. Noter les points incertains ou controverses

    Fournis un rapport structure avec les sources et extraits cles.""",
    expected_output="""Rapport de recherche contenant:
    - Liste des concepts cles identifies
    - Pour chaque source: titre, extrait pertinent, score de pertinence
    - Points d'incertitude ou necessitant verification""",
    agent=researcher,
)

analysis_task = Task(
    description="""Analyse les documents fournis par le chercheur.

    Tu dois:
    1. Verifier la coherence entre les sources
    2. Identifier les informations confirmees par plusieurs sources
    3. Signaler les contradictions eventuelles
    4. Evaluer le niveau de confiance pour chaque information
    5. Organiser les informations par theme

    Produis une synthese structuree avec niveaux de confiance.""",
    expected_output="""Synthese analytique contenant:
    - Informations confirmees (haute confiance)
    - Informations uniques (confiance moyenne)
    - Contradictions identifiees
    - Recommandations pour la redaction""",
    agent=analyst,
    context=[research_task],  # Depend de la recherche
)

writing_task = Task(
    description="""Redige une reponse complete basee sur l'analyse.

    Tu dois:
    1. Utiliser les informations validees par l'analyste
    2. Structurer la reponse avec introduction, corps, conclusion
    3. Citer les sources entre crochets [Source X]
    4. Adapter le niveau de detail au contexte
    5. Indiquer les limites ou incertitudes si necessaire

    La reponse doit etre professionnelle et bien formatee.""",
    expected_output="""Reponse structuree de 300-500 mots contenant:
    - Introduction contextualisant le sujet
    - Corps avec informations cles et citations
    - Conclusion synthetique
    - Liste des sources utilisees""",
    agent=writer,
    context=[research_task, analysis_task],
)

# 3. Creation de la Crew
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,  # Execution sequentielle
    verbose=True,
)

# 4. Execution
result = crew.kickoff(inputs={"topic": "Qu'est-ce que le RAG et comment l'implementer?"})
print(result)

Integration RAG avec CrewAI

Outils de recherche personnalises

DEVELOPERpython
from crewai_tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field
import json

class SearchInput(BaseModel):
    """Schema d'entree pour la recherche."""
    query: str = Field(description="La requete de recherche")
    top_k: int = Field(default=5, description="Nombre de resultats")

class VectorSearchTool(BaseTool):
    """Outil de recherche vectorielle pour CrewAI."""

    name: str = "search_knowledge_base"
    description: str = """Recherche des documents dans la base de connaissances.
    Utilise cet outil pour trouver des informations pertinentes sur un sujet."""
    args_schema: Type[BaseModel] = SearchInput

    def __init__(self, vector_store):
        super().__init__()
        self._vector_store = vector_store

    def _run(self, query: str, top_k: int = 5) -> str:
        """Execute la recherche."""
        results = self._vector_store.similarity_search(query, k=top_k)

        if not results:
            return "Aucun document pertinent trouve pour cette requete."

        output = "Documents trouves:\n\n"
        for i, doc in enumerate(results, 1):
            output += f"[Source {i}]\n"
            output += f"Contenu: {doc.page_content[:500]}...\n"
            if doc.metadata:
                output += f"Metadata: {doc.metadata}\n"
            output += "\n"

        return output


class FactVerificationTool(BaseTool):
    """Outil de verification des faits."""

    name: str = "verify_fact"
    description: str = """Verifie un fait contre les sources disponibles.
    Utilise cet outil pour confirmer ou infirmer une affirmation."""

    def __init__(self, vector_store):
        super().__init__()
        self._vector_store = vector_store

    def _run(self, claim: str) -> str:
        """Verifie une affirmation."""
        results = self._vector_store.similarity_search(claim, k=3)

        if not results:
            return json.dumps({
                "claim": claim,
                "status": "unverified",
                "confidence": 0,
                "reason": "Aucune source trouvee"
            })

        # Analyse simple de la coherence
        return json.dumps({
            "claim": claim,
            "status": "found_sources",
            "sources_count": len(results),
            "sources": [doc.page_content[:200] for doc in results]
        })


# Creation des outils
search_tool = VectorSearchTool(vector_store)
verify_tool = FactVerificationTool(vector_store)

# Agent avec outils RAG
rag_researcher = Agent(
    role="Expert RAG",
    goal="Rechercher et verifier des informations dans la base de connaissances",
    backstory="""Tu es un expert en recherche documentaire.
    Tu utilises systematiquement les outils de recherche pour
    trouver les informations les plus pertinentes et fiables.""",
    tools=[search_tool, verify_tool],
    llm=llm,
    verbose=True,
)

Crew RAG complete

DEVELOPERpython
from crewai import Agent, Task, Crew, Process

# Agents specialises pour RAG
query_analyst = Agent(
    role="Analyste de Requetes",
    goal="Decomposer et reformuler les questions pour optimiser la recherche",
    backstory="""Tu analyses les questions des utilisateurs pour identifier
    les concepts cles et generer des variantes de recherche efficaces.
    Tu detectes les ambiguites et les clarifies.""",
    llm=llm,
    verbose=True,
)

document_searcher = Agent(
    role="Chercheur Documentaire",
    goal="Trouver les documents les plus pertinents",
    backstory="""Tu es expert en recherche documentaire.
    Tu utilises differentes strategies de recherche pour
    maximiser la pertinence des resultats.""",
    tools=[search_tool],
    llm=llm,
    verbose=True,
)

fact_checker = Agent(
    role="Verificateur de Faits",
    goal="Verifier l'exactitude des informations",
    backstory="""Tu verifies chaque affirmation contre les sources.
    Tu identifies les incoherences et les potentielles erreurs.""",
    tools=[verify_tool],
    llm=llm,
    verbose=True,
)

response_generator = Agent(
    role="Generateur de Reponses",
    goal="Creer des reponses completes et bien sourcees",
    backstory="""Tu synthetises les informations trouvees
    en reponses claires, structurees et citant les sources.""",
    llm=llm,
    verbose=True,
)

# Tasks du pipeline RAG
analyze_query_task = Task(
    description="""Analyse cette question: {question}

    1. Identifie les concepts cles
    2. Genere 2-3 reformulations pour la recherche
    3. Identifie le type de question (factuelle, analytique, comparative)
    4. Liste les informations necessaires pour repondre""",
    expected_output="Analyse structuree de la question avec reformulations",
    agent=query_analyst,
)

search_documents_task = Task(
    description="""Recherche les documents pertinents.

    Utilise l'analyse de la question pour:
    1. Effectuer plusieurs recherches avec les reformulations
    2. Collecter au moins 5 documents pertinents
    3. Extraire les passages cles de chaque document""",
    expected_output="Liste de documents avec extraits pertinents",
    agent=document_searcher,
    context=[analyze_query_task],
)

verify_facts_task = Task(
    description="""Verifie les informations trouvees.

    Pour chaque information cle:
    1. Verifie la coherence entre sources
    2. Identifie les contradictions
    3. Attribue un niveau de confiance""",
    expected_output="Rapport de verification avec niveaux de confiance",
    agent=fact_checker,
    context=[search_documents_task],
)

generate_response_task = Task(
    description="""Genere la reponse finale.

    En utilisant les informations verifiees:
    1. Structure la reponse clairement
    2. Cite les sources [Source X]
    3. Indique les incertitudes si necessaire""",
    expected_output="Reponse finale structuree avec citations",
    agent=response_generator,
    context=[search_documents_task, verify_facts_task],
)

# Crew RAG
rag_crew = Crew(
    agents=[query_analyst, document_searcher, fact_checker, response_generator],
    tasks=[analyze_query_task, search_documents_task, verify_facts_task, generate_response_task],
    process=Process.sequential,
    verbose=True,
)

def answer_question(question: str) -> str:
    """Repond a une question via la crew RAG."""
    result = rag_crew.kickoff(inputs={"question": question})
    return result

Patterns avances

Pattern 1 : Crew hierarchique

Un manager supervise et delegue aux agents specialises :

DEVELOPERpython
from crewai import Agent, Task, Crew, Process

# Manager principal
project_manager = Agent(
    role="Chef de Projet RAG",
    goal="Coordonner l'equipe pour produire des reponses de qualite",
    backstory="""Tu supervises les equipes de recherche et redaction.
    Tu delegues les taches appropriees aux bons agents et
    valides la qualite des resultats.""",
    llm=llm,
    allow_delegation=True,  # Peut deleguer
    verbose=True,
)

# Tache principale
main_task = Task(
    description="""Reponds a cette question de maniere complete: {question}

    Tu dois coordonner l'equipe pour:
    1. Rechercher les informations pertinentes
    2. Verifier et analyser les resultats
    3. Produire une reponse de qualite

    Delegue les taches aux agents specialises.""",
    expected_output="Reponse complete, verifiee et bien structuree",
    agent=project_manager,
)

# Crew hierarchique
hierarchical_crew = Crew(
    agents=[project_manager, document_searcher, fact_checker, response_generator],
    tasks=[main_task],
    process=Process.hierarchical,  # Mode hierarchique
    manager_agent=project_manager,
    verbose=True,
)

Pattern 2 : Crew avec memoire

Agents qui maintiennent le contexte entre les requetes :

DEVELOPERpython
from crewai import Agent, Task, Crew
from crewai.memory import ShortTermMemory, LongTermMemory, EntityMemory

# Configuration memoire
short_memory = ShortTermMemory()
long_memory = LongTermMemory()
entity_memory = EntityMemory()

# Agent avec memoire
memory_researcher = Agent(
    role="Chercheur avec Memoire",
    goal="Rechercher en utilisant le contexte des interactions precedentes",
    backstory="""Tu te souviens des recherches precedentes
    et utilises ce contexte pour ameliorer tes resultats.
    Tu evites de repeter les memes recherches.""",
    llm=llm,
    memory=True,
    verbose=True,
)

# Crew avec memoire partagee
memory_crew = Crew(
    agents=[memory_researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    memory=True,
    short_term_memory=short_memory,
    long_term_memory=long_memory,
    entity_memory=entity_memory,
    verbose=True,
)

Pattern 3 : Crew conditionnelle

Execution conditionnelle basee sur les resultats :

DEVELOPERpython
from crewai import Agent, Task, Crew, Process

class ConditionalRAGCrew:
    """Crew avec logique conditionnelle."""

    def __init__(self, llm, vector_store):
        self.llm = llm
        self.vector_store = vector_store
        self._setup_agents()
        self._setup_tasks()

    def _setup_agents(self):
        self.classifier = Agent(
            role="Classificateur de Questions",
            goal="Determiner le type de question et la strategie",
            backstory="Tu analyses les questions pour orienter le traitement.",
            llm=self.llm,
        )

        self.simple_responder = Agent(
            role="Repondeur Simple",
            goal="Repondre aux questions factuelles simples",
            backstory="Tu reponds rapidement aux questions directes.",
            llm=self.llm,
        )

        self.complex_responder = Agent(
            role="Repondeur Complexe",
            goal="Traiter les questions necessitant une analyse approfondie",
            backstory="Tu geres les questions complexes avec rigueur.",
            llm=self.llm,
        )

    def _setup_tasks(self):
        self.classify_task = Task(
            description="""Classifie cette question: {question}

            Types possibles:
            - simple: Question factuelle directe
            - complex: Necessite recherche et analyse
            - comparative: Compare plusieurs elements

            Reponds uniquement avec le type.""",
            expected_output="Type de question (un mot)",
            agent=self.classifier,
        )

    def process(self, question: str) -> dict:
        """Traite une question avec logique conditionnelle."""
        # Phase 1: Classification
        classify_crew = Crew(
            agents=[self.classifier],
            tasks=[self.classify_task],
            process=Process.sequential,
        )
        classification = classify_crew.kickoff(inputs={"question": question})

        question_type = classification.strip().lower()

        # Phase 2: Traitement conditionnel
        if question_type == "simple":
            return self._process_simple(question)
        else:
            return self._process_complex(question)

    def _process_simple(self, question: str) -> dict:
        simple_task = Task(
            description=f"Reponds directement a: {question}",
            expected_output="Reponse concise et factuelle",
            agent=self.simple_responder,
        )
        crew = Crew(agents=[self.simple_responder], tasks=[simple_task])
        return {"type": "simple", "response": crew.kickoff()}

    def _process_complex(self, question: str) -> dict:
        complex_task = Task(
            description=f"Analyse et reponds en detail a: {question}",
            expected_output="Reponse detaillee avec sources",
            agent=self.complex_responder,
        )
        crew = Crew(agents=[self.complex_responder], tasks=[complex_task])
        return {"type": "complex", "response": crew.kickoff()}

Gestion des erreurs

DEVELOPERpython
from crewai import Agent, Task, Crew
from typing import Optional
import logging

logger = logging.getLogger(__name__)

class ResilientCrew:
    """Crew avec gestion robuste des erreurs."""

    def __init__(self, crew: Crew, max_retries: int = 3):
        self.crew = crew
        self.max_retries = max_retries

    def run_with_retry(self, inputs: dict) -> Optional[str]:
        """Execute la crew avec retry automatique."""
        for attempt in range(self.max_retries):
            try:
                result = self.crew.kickoff(inputs=inputs)
                return result

            except Exception as e:
                logger.warning(f"Tentative {attempt + 1} echouee: {e}")

                if attempt == self.max_retries - 1:
                    logger.error(f"Toutes les tentatives ont echoue")
                    return self._fallback_response(inputs, str(e))

        return None

    def _fallback_response(self, inputs: dict, error: str) -> str:
        """Reponse de secours en cas d'echec."""
        return f"""
        Desole, je n'ai pas pu traiter completement votre demande.

        Question: {inputs.get('question', 'N/A')}
        Erreur: {error}

        Suggestions:
        - Reformulez votre question
        - Essayez une question plus specifique
        - Contactez le support si le probleme persiste
        """


# Callback pour monitoring
def task_callback(task_output):
    """Callback appele apres chaque tache."""
    logger.info(f"Tache terminee: {task_output.description[:50]}...")
    logger.info(f"Longueur output: {len(str(task_output.raw))}")


# Crew avec callbacks
monitored_crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,
    task_callback=task_callback,
    verbose=True,
)

Performance et optimisation

DEVELOPERpython
from crewai import Crew, Process
import asyncio
from concurrent.futures import ThreadPoolExecutor

class OptimizedCrew:
    """Crew optimisee pour la performance."""

    def __init__(self, crew: Crew, cache_enabled: bool = True):
        self.crew = crew
        self.cache = {} if cache_enabled else None

    def run(self, inputs: dict) -> str:
        # Verifier le cache
        cache_key = self._get_cache_key(inputs)
        if self.cache and cache_key in self.cache:
            return self.cache[cache_key]

        # Executer
        result = self.crew.kickoff(inputs=inputs)

        # Mettre en cache
        if self.cache:
            self.cache[cache_key] = result

        return result

    def _get_cache_key(self, inputs: dict) -> str:
        """Genere une cle de cache."""
        import hashlib
        import json
        content = json.dumps(inputs, sort_keys=True)
        return hashlib.md5(content.encode()).hexdigest()


# Execution parallele de plusieurs crews
async def run_parallel_crews(crews_inputs: list) -> list:
    """Execute plusieurs crews en parallele."""
    async def run_crew(crew, inputs):
        loop = asyncio.get_event_loop()
        with ThreadPoolExecutor() as executor:
            result = await loop.run_in_executor(
                executor, crew.kickoff, inputs
            )
        return result

    tasks = [run_crew(c, i) for c, i in crews_inputs]
    return await asyncio.gather(*tasks)

Couts et metriques

Bonnes pratiques

1. Roles clairs et specifiques

DEVELOPERpython
# BON - Role precis avec contexte
Agent(
    role="Verificateur de Prix E-commerce",
    goal="Verifier les prix des produits contre les catalogues officiels",
    backstory="Expert e-commerce avec 5 ans d'experience en verification de donnees..."
)

# MAUVAIS - Role trop generique
Agent(
    role="Assistant",
    goal="Aider",
    backstory="Tu aides."
)

2. Backstory detaillee

DEVELOPERpython
# BON - Contexte riche qui guide le comportement
backstory="""Tu es un analyste senior en cybersecurite avec 8 ans d'experience.
Tu as travaille pour des entreprises du Fortune 500.
Tu excelles dans l'identification des vulnerabilites et la proposition de solutions.
Tu communiques de maniere claire et actionnable."""

# MAUVAIS - Trop court, ne guide pas
backstory="Tu es un expert en securite."

3. Tasks avec outputs attendus clairs

DEVELOPERpython
# BON - Output specifique et structure
Task(
    description="Analyse les logs de securite...",
    expected_output="""Rapport JSON contenant:
    - vulnerabilities: liste des vulnerabilites trouvees
    - severity: niveau de severite (low/medium/high/critical)
    - recommendations: actions correctives
    - confidence: niveau de confiance (0-1)"""
)

# MAUVAIS - Output vague
Task(
    description="Analyse les logs...",
    expected_output="Un rapport"
)

Conclusion

CrewAI offre une abstraction intuitive pour construire des systemes RAG multi-agents. L'approche "equipe" avec roles, taches et delegation rend le code lisible et maintenable.

Points cles :

• Definissez des roles precis avec des backstories detaillees
• Utilisez des outils personnalises pour l'integration RAG
• Implementez des patterns hierarchiques pour les workflows complexes
• Ajoutez de la memoire pour les conversations contextuelles

Pour aller plus loin

• LangGraph : Workflows RAG - Approche graphe
• AutoGen : Multi-agents Microsoft - Approche conversationnelle
• Function calling RAG - Actions dans le RAG

Besoin d'equipes d'agents? Ailog utilise des architectures inspirees de CrewAI pour orchestrer des agents specialises. Deploiement simple, collaboration efficace.