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 : Si une récupération est nécessaire Quoi rechercher (formulation de requêtes optimales) Où chercher (sélection de sources) Quand s'arrêter (critères de suffisance) Comment combiner les résultats (synthèse multi-sources)

RAG Classique vs Agentic RAG

| Aspect | RAG Classique | Agentic RAG | |--------|---------------|-------------| | Flux | Linéaire (requête → récupération → génération) | Itératif et adaptatif | | Décision de récupération | Toujours (systématique) | Conditionnelle (quand nécessaire) | | Formulation de requête | Requête utilisateur directe | Requêtes optimisées par l'agent | | Sources | Fixe (une base de connaissances) | Multiple et dynamique | | Validation | Aucune | Auto-vérification et correction | | Raisonnement | Single-hop | Multi-hop avec chaînage | | Complexité | Faible | Élevée | | Cas d'usage | Questions factuelles simples | Tâches complexes et recherche |

Pourquoi l'Agentic RAG ?

Limites du RAG classique : Questions complexes : "Compare les stratégies de pricing de nos 3 principaux concurrents et recommande une position" nécessite plusieurs recherches et synthèse. Informations incomplètes : Si la première récupération ne suffit pas, le RAG classique ne peut pas chercher davantage. Requêtes ambiguës : L'agent peut clarifier ou reformuler avant de chercher. Hallucinations non détectées : L'agent peut vérifier ses propres réponses contre les sources. 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 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.

`python 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 ` Raisonneur (Reasoner)

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

`python 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: Les documents répondent-ils à la question ? Y a-t-il des informations manquantes ? Y a-t-il des contradictions ? 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 ` Gestionnaire de Mémoire (Memory Manager)

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

`python 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 ` Exécuteur (Executor)

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

`python 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 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 :

`python 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." ` Pattern Plan-and-Execute

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

`python 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) ` Pattern Self-RAG

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

`python 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 ` Pattern Corrective RAG (CRAG)

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

`python 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

`python 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: Commence toujours par réfléchir à ce dont tu as besoin Utilise knowledge_search pour les informations internes Utilise web_search pour les informations récentes ou externes Vérifie tes informations avant de conclure 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

`python 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

`python 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: FACTUALITÉ: Chaque affirmation est-elle supportée par les sources? (oui/non/partiel) COMPLÉTUDE: La réponse couvre-t-elle tous les aspects de la question? (oui/non) COHÉRENCE: La réponse est-elle logiquement cohérente? (oui/non) 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: Élimine les hallucinations S'appuie uniquement sur les sources Reste complète et utile """

return self.llm.generate(prompt) `

Cas d'Usage Avancés Assistant de Recherche Multi-Documents

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

`python 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 ` Agent de Due Diligence

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

`python 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) ` Agent de Support Client Intelligent

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

`python 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 Gestion des Tokens et Coûts

`python 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) ` Parallélisation des Recherches

`python 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) ` Caching Intelligent

`python 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 ` Gestion des Erreurs et Fallbacks

`python 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 Taux de résolution : Pourcentage de requêtes résolues sans intervention humaine Nombre d'étapes : Efficacité du raisonnement (moins = mieux) Précision des récupérations : Pertinence des documents trouvés Fidélité : Réponses basées sur les sources vs hallucinations Latence end-to-end : Temps total de résolution

Framework d'Évaluation

`python 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 : Pensez agent, pas pipeline : L'agent décide dynamiquement de ses actions Modularité : Séparez planification, exécution et évaluation Validation continue : L'agent doit critiquer ses propres résultats Optimisation : Parallélisez, cachez, et gérez les tokens 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