Agentic RAG : Construire des Agents IA avec Récupération Dynamique de Connaissances

TL;DR

• Agentic RAG combine la puissance des agents IA autonomes avec la récupération de connaissances RAG
• Les agents peuvent décider dynamiquement quand et quoi récupérer, contrairement au RAG classique
• Architecture modulaire : planificateur, récupérateur, raisonneur, exécuteur
• Patterns clés : ReAct, Plan-and-Execute, Self-RAG, Corrective RAG
• Cas d'usage : assistants de recherche, automatisation complexe, analyse multi-documents
• Testez maintenant : Construisez votre agent RAG sur Ailog

Introduction : Au-delà du RAG Traditionnel

Le RAG (Retrieval-Augmented Generation) classique suit un pipeline linéaire : requête → récupération → génération. Cette approche fonctionne bien pour des questions simples, mais atteint ses limites face à des tâches complexes nécessitant :

• Plusieurs étapes de raisonnement
• La combinaison d'informations de sources multiples
• Des décisions dynamiques sur ce qu'il faut rechercher
• La validation et correction des informations récupérées

L'Agentic RAG répond à ces défis en donnant à l'IA la capacité de planifier, exécuter et itérer de manière autonome. L'agent devient un orchestrateur intelligent qui décide quand récupérer, quoi chercher, et comment combiner les informations pour accomplir des tâches complexes.

Qu'est-ce que l'Agentic RAG ?

Définition

L'Agentic RAG est une architecture où un agent IA autonome utilise la récupération de connaissances comme l'un de ses outils, parmi d'autres, pour accomplir des tâches. Contrairement au RAG traditionnel où la récupération est systématique, l'agent décide dynamiquement :

1. Si une récupération est nécessaire
2. Quoi rechercher (formulation de requêtes optimales)
3. Où chercher (sélection de sources)
4. Quand s'arrêter (critères de suffisance)
5. Comment combiner les résultats (synthèse multi-sources)

RAG Classique vs Agentic RAG

Pourquoi l'Agentic RAG ?

Limites du RAG classique :

1. Questions complexes : "Compare les stratégies de pricing de nos 3 principaux concurrents et recommande une position" nécessite plusieurs recherches et synthèse.

2. Informations incomplètes : Si la première récupération ne suffit pas, le RAG classique ne peut pas chercher davantage.

3. Requêtes ambiguës : L'agent peut clarifier ou reformuler avant de chercher.

4. Hallucinations non détectées : L'agent peut vérifier ses propres réponses contre les sources.

5. Tâches multi-étapes : Réserver un voyage nécessite rechercher vols, hôtels, puis combiner et valider.

Architecture de l'Agentic RAG

Vue d'Ensemble

┌─────────────────────────────────────────────────────────────────┐
│                         AGENT CONTROLLER                         │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────────┐  │
│  │  Planificateur│  │  Raisonneur  │  │  Gestionnaire Mémoire │  │
│  └──────┬───────┘  └──────┬───────┘  └──────────┬───────────┘  │
│         │                 │                      │              │
│         └────────────┬────┴──────────────────────┘              │
│                      │                                           │
│              ┌───────▼───────┐                                   │
│              │   Exécuteur   │                                   │
│              └───────┬───────┘                                   │
└──────────────────────┼──────────────────────────────────────────┘
                       │
        ┌──────────────┼──────────────┐
        │              │              │
   ┌────▼────┐   ┌────▼────┐   ┌────▼────┐
   │  Outil  │   │  Outil  │   │  Outil  │
   │Retrieval│   │ Calcul  │   │  API    │
   └─────────┘   └─────────┘   └─────────┘

Composants Clés

1. Planificateur (Planner)

Le planificateur décompose les tâches complexes en sous-tâches gérables. Il maintient un plan d'exécution qui peut être révisé dynamiquement.

DEVELOPERpython
class Planner:
    def __init__(self, llm):
        self.llm = llm

    def create_plan(self, task: str, context: dict) -> List[Step]:
        """Décompose une tâche en étapes exécutables."""
        prompt = f"""
        Tâche: {task}
        Contexte disponible: {context}

        Décompose cette tâche en étapes claires et ordonnées.
        Pour chaque étape, indique:
        - L'action à effectuer
        - Les outils nécessaires
        - Les dépendances avec d'autres étapes
        """

        plan = self.llm.generate(prompt)
        return self.parse_plan(plan)

    def revise_plan(self, plan: List[Step], feedback: str) -> List[Step]:
        """Révise le plan en fonction des résultats intermédiaires."""
        # Adapte le plan si des informations manquent ou changent
        pass

2. Raisonneur (Reasoner)

Le raisonneur analyse les informations récupérées, identifie les lacunes, et décide des prochaines actions.

DEVELOPERpython
class Reasoner:
    def __init__(self, llm):
        self.llm = llm

    def analyze_retrieval(self, query: str, documents: List[Document]) -> Analysis:
        """Analyse si les documents récupérés sont suffisants."""
        prompt = f"""
        Question: {query}
        Documents récupérés: {documents}

        Analyse:
        1. Les documents répondent-ils à la question ?
        2. Y a-t-il des informations manquantes ?
        3. Y a-t-il des contradictions ?
        4. Quelle confiance accordes-tu aux informations ?

        Décision: [SUFFICIENT | NEED_MORE | REFORMULATE | ESCALATE]
        """
        return self.llm.generate(prompt)

    def synthesize(self, query: str, all_results: List[RetrievalResult]) -> str:
        """Synthétise les informations de plusieurs récupérations."""
        pass

3. Gestionnaire de Mémoire (Memory Manager)

Maintient le contexte conversationnel et les résultats intermédiaires.

DEVELOPERpython
class MemoryManager:
    def __init__(self):
        self.short_term = []  # Conversation actuelle
        self.working_memory = {}  # Résultats intermédiaires
        self.episodic = []  # Historique des actions

    def add_to_working_memory(self, key: str, value: any):
        """Stocke un résultat intermédiaire."""
        self.working_memory[key] = {
            "value": value,
            "timestamp": datetime.now(),
            "source": "retrieval"  # ou "computation", "user"
        }

    def get_relevant_context(self, query: str) -> dict:
        """Récupère le contexte pertinent pour une requête."""
        # Combine mémoire court terme et résultats intermédiaires
        pass

4. Exécuteur (Executor)

Orchestre l'exécution des outils selon le plan.

DEVELOPERpython
class Executor:
    def __init__(self, tools: Dict[str, Tool]):
        self.tools = tools

    async def execute_step(self, step: Step) -> StepResult:
        """Exécute une étape du plan."""
        tool = self.tools[step.tool_name]
        result = await tool.execute(step.parameters)

        return StepResult(
            step=step,
            result=result,
            success=result.is_valid(),
            metadata={"latency": result.latency}
        )

    async def execute_plan(self, plan: List[Step]) -> ExecutionResult:
        """Exécute un plan complet avec gestion des erreurs."""
        results = []
        for step in plan:
            result = await self.execute_step(step)
            results.append(result)

            if not result.success and step.is_critical:
                # Déclenche une révision du plan
                break

        return ExecutionResult(results=results)

Patterns d'Agentic RAG

1. Pattern ReAct (Reasoning + Acting)

ReAct alterne entre réflexion et action. L'agent pense à voix haute avant chaque action.

Thought: Je dois trouver les revenus Q3 2024 de l'entreprise X.
Action: search_documents("revenus Q3 2024 entreprise X")
Observation: Document trouvé: Rapport financier Q3 2024, revenus = 45M€
Thought: J'ai les revenus, maintenant je dois les comparer au Q3 2023.
Action: search_documents("revenus Q3 2023 entreprise X")
Observation: Document trouvé: Rapport financier Q3 2023, revenus = 38M€
Thought: Je peux maintenant calculer la croissance.
Action: calculate((45-38)/38 * 100)
Observation: Résultat: 18.4%
Thought: J'ai toutes les informations pour répondre.
Final Answer: L'entreprise X a réalisé 45M€ de revenus au Q3 2024, soit une croissance de 18.4% par rapport au Q3 2023 (38M€).

Implémentation :

DEVELOPERpython
class ReActAgent:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
        self.max_iterations = 10

    def run(self, query: str) -> str:
        history = []

        for i in range(self.max_iterations):
            # Génère la prochaine pensée et action
            prompt = self.build_prompt(query, history)
            response = self.llm.generate(prompt)

            thought, action = self.parse_response(response)
            history.append({"thought": thought, "action": action})

            # Vérifie si c'est une réponse finale
            if action.type == "final_answer":
                return action.content

            # Exécute l'action
            observation = self.execute_action(action)
            history.append({"observation": observation})

        return "Impossible de trouver une réponse satisfaisante."

2. Pattern Plan-and-Execute

Sépare la planification de l'exécution pour les tâches complexes.

DEVELOPERpython
class PlanAndExecuteAgent:
    def __init__(self, planner, executor, replanner):
        self.planner = planner
        self.executor = executor
        self.replanner = replanner

    async def run(self, task: str) -> str:
        # Phase 1: Planification initiale
        plan = self.planner.create_plan(task)

        results = []
        for step in plan:
            # Phase 2: Exécution
            result = await self.executor.execute_step(step)
            results.append(result)

            # Phase 3: Replanification si nécessaire
            if result.requires_replan:
                remaining_steps = plan[plan.index(step)+1:]
                plan = self.replanner.revise(
                    original_task=task,
                    completed=results,
                    remaining=remaining_steps,
                    feedback=result.feedback
                )

        return self.synthesize_results(results)

3. Pattern Self-RAG

L'agent évalue et critique ses propres récupérations et générations.

DEVELOPERpython
class SelfRAGAgent:
    def __init__(self, llm, retriever):
        self.llm = llm
        self.retriever = retriever

    def run(self, query: str) -> str:
        # Étape 1: Décider si la récupération est nécessaire
        need_retrieval = self.assess_retrieval_need(query)

        if need_retrieval:
            # Étape 2: Récupérer
            documents = self.retriever.search(query)

            # Étape 3: Critiquer la pertinence
            relevant_docs = self.critique_relevance(query, documents)

            if not relevant_docs:
                # Reformuler et réessayer
                new_query = self.reformulate_query(query)
                documents = self.retriever.search(new_query)
                relevant_docs = self.critique_relevance(query, documents)

        # Étape 4: Générer la réponse
        response = self.generate_response(query, relevant_docs)

        # Étape 5: Critiquer la réponse
        is_supported = self.critique_support(response, relevant_docs)
        is_useful = self.critique_usefulness(response, query)

        if not is_supported or not is_useful:
            # Régénérer avec feedback
            response = self.regenerate_with_feedback(
                query, relevant_docs,
                support_feedback=is_supported,
                usefulness_feedback=is_useful
            )

        return response

4. Pattern Corrective RAG (CRAG)

Évalue la qualité des documents récupérés et prend des actions correctives.

DEVELOPERpython
class CorrectiveRAGAgent:
    def __init__(self, llm, retriever, web_search):
        self.llm = llm
        self.retriever = retriever
        self.web_search = web_search

    def run(self, query: str) -> str:
        # Récupération initiale
        documents = self.retriever.search(query)

        # Évaluation de la qualité
        relevance_scores = self.evaluate_relevance(query, documents)

        # Classification des documents
        correct_docs = [d for d, s in zip(documents, relevance_scores) if s > 0.7]
        ambiguous_docs = [d for d, s in zip(documents, relevance_scores) if 0.3 < s <= 0.7]
        incorrect_docs = [d for d, s in zip(documents, relevance_scores) if s <= 0.3]

        # Actions correctives selon le cas
        if len(correct_docs) >= 2:
            # Cas: Documents suffisants
            final_docs = correct_docs
        elif len(correct_docs) + len(ambiguous_docs) >= 2:
            # Cas: Besoin de raffiner les ambigus
            refined = self.refine_ambiguous(query, ambiguous_docs)
            final_docs = correct_docs + refined
        else:
            # Cas: Besoin de recherche externe
            web_results = self.web_search.search(query)
            final_docs = correct_docs + self.process_web_results(web_results)

        # Génération avec les documents corrigés
        return self.generate_response(query, final_docs)

Implémentation Pratique

Configuration d'un Agent RAG avec LangChain

DEVELOPERpython
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.tools import Tool
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

# Définir les outils
def search_knowledge_base(query: str) -> str:
    """Recherche dans la base de connaissances interne."""
    # Implémentation de la recherche vectorielle
    results = vector_store.similarity_search(query, k=5)
    return "\n".join([doc.page_content for doc in results])

def search_web(query: str) -> str:
    """Recherche sur le web pour des informations récentes."""
    # Implémentation de la recherche web
    pass

def calculate(expression: str) -> str:
    """Effectue un calcul mathématique."""
    return str(eval(expression))

tools = [
    Tool(
        name="knowledge_search",
        func=search_knowledge_base,
        description="Recherche dans la documentation interne et les bases de connaissances. Utiliser pour des informations spécifiques à l'entreprise."
    ),
    Tool(
        name="web_search",
        func=search_web,
        description="Recherche sur le web. Utiliser pour des informations récentes ou publiques."
    ),
    Tool(
        name="calculator",
        func=calculate,
        description="Effectue des calculs mathématiques. Input: expression mathématique valide."
    )
]

# Créer le prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", """Tu es un assistant de recherche expert. Tu utilises tes outils de manière judicieuse pour répondre aux questions.

Règles:
1. Commence toujours par réfléchir à ce dont tu as besoin
2. Utilise knowledge_search pour les informations internes
3. Utilise web_search pour les informations récentes ou externes
4. Vérifie tes informations avant de conclure
5. Cite tes sources dans ta réponse finale"""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

# Créer l'agent
llm = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0)
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Exécuter
response = agent_executor.invoke({
    "input": "Compare nos ventes Q3 2024 avec la moyenne du marché"
})

Gestion Multi-Sources avec Routage

DEVELOPERpython
class MultiSourceRouter:
    """Route les requêtes vers les sources appropriées."""

    def __init__(self, sources: Dict[str, VectorStore], llm):
        self.sources = sources
        self.llm = llm

    def route(self, query: str) -> List[str]:
        """Détermine quelles sources interroger."""
        prompt = f"""
        Requête: {query}

        Sources disponibles:
        - documentation_technique: Docs techniques, APIs, architecture
        - base_clients: Informations clients, contrats, historique
        - finance: Rapports financiers, budgets, prévisions
        - rh: Politiques RH, organigramme, procédures
        - produits: Catalogue produits, pricing, specs

        Quelles sources sont pertinentes pour cette requête?
        Réponds avec une liste JSON: ["source1", "source2"]
        """

        response = self.llm.generate(prompt)
        return json.loads(response)

    async def search_all(self, query: str) -> Dict[str, List[Document]]:
        """Recherche parallèle dans toutes les sources pertinentes."""
        relevant_sources = self.route(query)

        tasks = [
            self.search_source(source, query)
            for source in relevant_sources
        ]

        results = await asyncio.gather(*tasks)
        return dict(zip(relevant_sources, results))

Validation et Auto-Correction

DEVELOPERpython
class ResponseValidator:
    """Valide et corrige les réponses générées."""

    def __init__(self, llm):
        self.llm = llm

    def validate(self, query: str, response: str, sources: List[Document]) -> ValidationResult:
        prompt = f"""
        Question: {query}
        Réponse générée: {response}
        Sources utilisées: {[doc.page_content for doc in sources]}

        Évalue cette réponse:
        1. FACTUALITÉ: Chaque affirmation est-elle supportée par les sources? (oui/non/partiel)
        2. COMPLÉTUDE: La réponse couvre-t-elle tous les aspects de la question? (oui/non)
        3. COHÉRENCE: La réponse est-elle logiquement cohérente? (oui/non)
        4. HALLUCINATIONS: Y a-t-il des informations non présentes dans les sources? (liste)

        Format JSON:
        {{
            "factuality": "oui|non|partiel",
            "completeness": "oui|non",
            "coherence": "oui|non",
            "hallucinations": ["...", "..."],
            "confidence": 0.0-1.0,
            "corrections_needed": ["...", "..."]
        }}
        """

        result = self.llm.generate(prompt)
        return ValidationResult.from_json(result)

    def correct(self, query: str, response: str, validation: ValidationResult, sources: List[Document]) -> str:
        """Corrige la réponse en fonction de la validation."""
        if validation.confidence > 0.9:
            return response

        prompt = f"""
        Réponse originale: {response}
        Problèmes identifiés: {validation.corrections_needed}
        Hallucinations: {validation.hallucinations}
        Sources correctes: {[doc.page_content for doc in sources]}

        Génère une réponse corrigée qui:
        1. Élimine les hallucinations
        2. S'appuie uniquement sur les sources
        3. Reste complète et utile
        """

        return self.llm.generate(prompt)

Cas d'Usage Avancés

1. Assistant de Recherche Multi-Documents

Analyse et synthétise des informations provenant de nombreux documents.

DEVELOPERpython
class ResearchAssistant:
    """Assistant de recherche capable d'analyser plusieurs documents."""

    async def research(self, topic: str, depth: str = "comprehensive") -> ResearchReport:
        # Phase 1: Exploration initiale
        initial_results = await self.broad_search(topic)

        # Phase 2: Identification des sous-thèmes
        subtopics = self.identify_subtopics(topic, initial_results)

        # Phase 3: Recherche approfondie par sous-thème
        detailed_results = {}
        for subtopic in subtopics:
            results = await self.deep_search(subtopic)
            detailed_results[subtopic] = results

        # Phase 4: Identification des contradictions
        contradictions = self.find_contradictions(detailed_results)

        # Phase 5: Synthèse
        report = self.synthesize_report(
            topic=topic,
            subtopics=subtopics,
            results=detailed_results,
            contradictions=contradictions
        )

        return report

2. Agent de Due Diligence

Automatise l'analyse approfondie pour des décisions business.

DEVELOPERpython
class DueDiligenceAgent:
    """Agent pour l'analyse due diligence automatisée."""

    def analyze_company(self, company_name: str) -> DueDiligenceReport:
        sections = [
            ("financial", self.analyze_financials),
            ("legal", self.analyze_legal),
            ("market", self.analyze_market_position),
            ("team", self.analyze_leadership),
            ("tech", self.analyze_technology),
            ("risks", self.identify_risks)
        ]

        results = {}
        for section_name, analyzer in sections:
            results[section_name] = analyzer(company_name)

        # Synthèse et scoring
        return self.compile_report(company_name, results)

3. Agent de Support Client Intelligent

Résout des problèmes complexes en consultant plusieurs sources.

DEVELOPERpython
class SupportAgent:
    """Agent de support client avec résolution multi-étapes."""

    async def handle_ticket(self, ticket: SupportTicket) -> Resolution:
        # Comprendre le problème
        problem_analysis = self.analyze_problem(ticket)

        # Rechercher des solutions
        kb_results = await self.search_knowledge_base(problem_analysis.keywords)
        past_tickets = await self.search_similar_tickets(problem_analysis)

        # Évaluer les solutions potentielles
        solutions = self.evaluate_solutions(kb_results, past_tickets)

        if solutions.best.confidence > 0.8:
            return self.generate_resolution(solutions.best)
        else:
            # Escalader avec contexte enrichi
            return self.escalate_with_context(ticket, problem_analysis, solutions)

Optimisation et Bonnes Pratiques

1. Gestion des Tokens et Coûts

DEVELOPERpython
class TokenOptimizer:
    """Optimise l'utilisation des tokens dans les agents."""

    def __init__(self, max_tokens_per_step: int = 2000):
        self.max_tokens = max_tokens_per_step

    def compress_context(self, documents: List[Document], query: str) -> str:
        """Compresse le contexte pour respecter les limites."""
        # Trier par pertinence
        scored = [(doc, self.relevance_score(doc, query)) for doc in documents]
        scored.sort(key=lambda x: x[1], reverse=True)

        # Sélectionner jusqu'à la limite
        selected = []
        token_count = 0
        for doc, score in scored:
            doc_tokens = self.count_tokens(doc.page_content)
            if token_count + doc_tokens <= self.max_tokens:
                selected.append(doc.page_content)
                token_count += doc_tokens

        return "\n---\n".join(selected)

2. Parallélisation des Recherches

DEVELOPERpython
async def parallel_search(queries: List[str], retrievers: List[Retriever]) -> Dict:
    """Exécute plusieurs recherches en parallèle."""
    tasks = []
    for query in queries:
        for retriever in retrievers:
            tasks.append(retriever.search(query))

    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Regrouper et dédupliquer les résultats
    return deduplicate_results(results)

3. Caching Intelligent

DEVELOPERpython
class AgentCache:
    """Cache intelligent pour les résultats d'agents."""

    def __init__(self, ttl: int = 3600):
        self.cache = {}
        self.ttl = ttl

    def get_or_compute(self, key: str, compute_fn: Callable) -> Any:
        # Vérifier le cache
        if key in self.cache:
            entry = self.cache[key]
            if time.time() - entry["timestamp"] < self.ttl:
                return entry["value"]

        # Calculer et mettre en cache
        result = compute_fn()
        self.cache[key] = {
            "value": result,
            "timestamp": time.time()
        }
        return result

4. Gestion des Erreurs et Fallbacks

DEVELOPERpython
class ResilientAgent:
    """Agent avec gestion robuste des erreurs."""

    async def execute_with_fallback(self, action: Action) -> Result:
        strategies = [
            (action.primary_tool, action.params),
            (action.fallback_tool, action.params),
            (self.web_search, {"query": action.query}),
            (self.ask_user, {"question": f"Je n'ai pas pu trouver: {action.query}"})
        ]

        for tool, params in strategies:
            try:
                result = await asyncio.wait_for(
                    tool.execute(params),
                    timeout=30.0
                )
                if result.is_valid():
                    return result
            except Exception as e:
                self.log_error(e, tool, params)
                continue

        return Result.failure("Toutes les stratégies ont échoué")

Évaluation des Agents RAG

Métriques Clés

1. Taux de résolution : Pourcentage de requêtes résolues sans intervention humaine
2. Nombre d'étapes : Efficacité du raisonnement (moins = mieux)
3. Précision des récupérations : Pertinence des documents trouvés
4. Fidélité : Réponses basées sur les sources vs hallucinations
5. Latence end-to-end : Temps total de résolution

Framework d'Évaluation

DEVELOPERpython
class AgentEvaluator:
    """Évalue les performances d'un agent RAG."""

    def evaluate(self, agent: Agent, test_cases: List[TestCase]) -> EvaluationReport:
        metrics = {
            "resolution_rate": [],
            "steps_count": [],
            "retrieval_precision": [],
            "faithfulness": [],
            "latency": []
        }

        for case in test_cases:
            start = time.time()
            result = agent.run(case.query)
            latency = time.time() - start

            metrics["latency"].append(latency)
            metrics["resolution_rate"].append(
                self.check_resolution(result, case.expected)
            )
            metrics["faithfulness"].append(
                self.check_faithfulness(result, agent.last_sources)
            )
            # ... autres métriques

        return EvaluationReport(
            avg_resolution_rate=np.mean(metrics["resolution_rate"]),
            avg_latency=np.mean(metrics["latency"]),
            # ...
        )

Conclusion

L'Agentic RAG représente l'évolution naturelle des systèmes RAG vers plus d'autonomie et d'intelligence. En combinant planification, raisonnement et récupération dynamique, ces agents peuvent résoudre des tâches complexes qui dépassent les capacités du RAG traditionnel.

Points clés à retenir :

1. Pensez agent, pas pipeline : L'agent décide dynamiquement de ses actions
2. Modularité : Séparez planification, exécution et évaluation
3. Validation continue : L'agent doit critiquer ses propres résultats
4. Optimisation : Parallélisez, cachez, et gérez les tokens
5. Résilience : Prévoyez des fallbacks et une gestion d'erreurs robuste

L'Agentic RAG ouvre la voie vers des assistants IA véritablement capables de recherche autonome, d'analyse complexe et de raisonnement multi-étapes. C'est le fondement des futurs systèmes d'IA capables de travailler en autonomie sur des tâches sophistiquées.

Ressources Complémentaires

• Documentation Ailog - Guides complets sur le RAG
• LangChain Agents - Framework d'agents
• Paper ReAct - Pattern ReAct original
• Paper Self-RAG - Self-Reflective RAG
• Paper CRAG - Corrective RAG