AnleitungExperte

Agentic RAG 2025: Aufbau autonomer KI-Agenten (Kompletter Leitfaden)

28. Januar 2025
25 Min. Lesezeit
Ailog Research Team

Kompletter Agentic RAG-Leitfaden: Architektur, Design Patterns, autonome Agenten mit dynamischem Retrieval, Multi-Tool-Orchestrierung. Mit Beispielen LangGraph und CrewAI.

TL;DR

  • Agentic RAG kombiniert die Leistungsfähigkeit autonomer IA-Agenten mit Retrieval-gestützter Wissensnutzung (RAG)
  • Agenten können dynamisch entscheiden, wann und was abgerufen werden muss, im Gegensatz zum klassischen RAG
  • Modulare Architektur : Planer, Retriever, Reasoner, Executor
  • Wichtige Patterns : ReAct, Plan-and-Execute, Self-RAG, Corrective RAG
  • Anwendungsfälle : Research-Assistenten, komplexe Automatisierung, Multi-Dokument-Analyse
  • Jetzt testen : Baue deinen RAG-Agenten auf Ailog

Introduction : Au-delà du RAG Traditionnel

Das klassische RAG (Retrieval-Augmented Generation) folgt einer linearen Pipeline: Anfrage → retrieval → Generierung. Dieser Ansatz funktioniert gut für einfache Fragen, stößt jedoch an seine Grenzen bei komplexen Aufgaben, die erfordern:

  • Mehrstufiges Reasoning
  • Die Kombination von Informationen aus mehreren Quellen
  • Dynamische Entscheidungen darüber, was abgerufen werden muss
  • Validierung und Korrektur der abgerufenen Informationen

Das Agentic RAG adressiert diese Herausforderungen, indem es der IA die Fähigkeit gibt, autonom zu planen, auszuführen und zu iterieren. Der Agent wird zu einem intelligenten Orchestrator, der entscheidet, wann er abruft, was er sucht und wie Informationen kombiniert werden, um komplexe Aufgaben zu lösen.

Qu'est-ce que l'Agentic RAG ?

Définition

Agentic RAG ist eine Architektur, in der ein autonomer IA-Agent retrieval als eines seiner Werkzeuge unter mehreren nutzt, um Aufgaben zu erfüllen. Im Gegensatz zum traditionellen RAG, bei dem retrieval systematisch erfolgt, entscheidet der Agent dynamisch:

  1. Si eine retrieval notwendig ist
  2. Quoi zu suchen (Formulierung optimaler Queries)
  3. zu suchen (Quellenauswahl)
  4. Quand aufzuhören (Kriterien für Suffizienz)
  5. Comment die Ergebnisse zu kombinieren (Multi-Source-Synthese)

RAG Classique vs Agentic RAG

AspectRAG ClassiqueAgentic RAG
FluxLinéaire (requête → récupération → génération)Itératif et adaptatif
Décision de récupérationToujours (systématique)Conditionnelle (quand nécessaire)
Formulation de requêteRequête utilisateur directeRequêtes optimisées par l'agent
SourcesFixe (une base de connaissances)Multiple et dynamique
ValidationAucuneAuto-vérification et correction
RaisonnementSingle-hopMulti-hop avec chaînage
ComplexitéFaibleÉlevée
Cas d'usageQuestions factuelles simplesTâches complexes et recherche

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.""" # Passt den Plan an, wenn Informationen fehlen oder sich ändern 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" # oder "computation", "user" } def get_relevant_context(self, query: str) -> dict: """Récupère le contexte pertinent pour une requête.""" # Kombiniert Kurzzeitgedächtnis und Zwischenergebnisse 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: # Löst eine Überarbeitung des Plans aus 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): # Generiert den nächsten Gedanken und die Aktion prompt = self.build_prompt(query, history) response = self.llm.generate(prompt) thought, action = self.parse_response(response) history.append({"thought": thought, "action": action}) # Prüft, ob es eine finale Antwort ist if action.type == "final_answer": return action.content # Führt die Aktion aus 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: Initiale Planung plan = self.planner.create_plan(task) results = [] for step in plan: # Phase 2: Ausführung result = await self.executor.execute_step(step) results.append(result) # Phase 3: Neoplanung falls nötig 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: # Schritt 1: Entscheiden, ob retrieval notwendig ist need_retrieval = self.assess_retrieval_need(query) if need_retrieval: # Schritt 2: Retrieval durchführen documents = self.retriever.search(query) # Schritt 3: Relevanz kritisch bewerten relevant_docs = self.critique_relevance(query, documents) if not relevant_docs: # Umformulieren und erneut versuchen new_query = self.reformulate_query(query) documents = self.retriever.search(new_query) relevant_docs = self.critique_relevance(query, documents) # Schritt 4: Antwort generieren response = self.generate_response(query, relevant_docs) # Schritt 5: Antwort kritisch bewerten is_supported = self.critique_support(response, relevant_docs) is_useful = self.critique_usefulness(response, query) if not is_supported or not is_useful: # Mit Feedback neu generieren 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: # Initiales Retrieval documents = self.retriever.search(query) # Qualitätsbewertung relevance_scores = self.evaluate_relevance(query, documents) # Klassifizierung der Dokumente 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: # Fall: Ausreichende Dokumente final_docs = correct_docs elif len(correct_docs) + len(ambiguous_docs) >= 2: # Fall: Ambiguës verfeinern refined = self.refine_ambiguous(query, ambiguous_docs) final_docs = correct_docs + refined else: # Fall: Recherche externe nécessaire 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.""" # Implementierung der vector-Suche 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.""" # Implementierung der Websuche 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: Erste Erkundung initial_results = await self.broad_search(topic) # Phase 2: Identifikation der Unterthemen subtopics = self.identify_subtopics(topic, initial_results) # Phase 3: Tiefgehende Recherche pro Unterthema detailed_results = {} for subtopic in subtopics: results = await self.deep_search(subtopic) detailed_results[subtopic] = results # Phase 4: Erkennung von Widersprüchen contradictions = self.find_contradictions(detailed_results) # Phase 5: Synthese 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: # Problem verstehen problem_analysis = self.analyze_problem(ticket) # Lösungen recherchieren kb_results = await self.search_knowledge_base(problem_analysis.keywords) past_tickets = await self.search_similar_tickets(problem_analysis) # Potentielle Lösungen bewerten solutions = self.evaluate_solutions(kb_results, past_tickets) if solutions.best.confidence > 0.8: return self.generate_resolution(solutions.best) else: # Mit erweitertem Kontext eskalieren 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.""" # Nach Relevanz sortieren scored = [(doc, self.relevance_score(doc, query)) for doc in documents] scored.sort(key=lambda x: x[1], reverse=True) # Bis zur Grenze auswählen 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) # Ergebnisse zusammenführen und deduplizieren 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: # Cache überprüfen if key in self.cache: entry = self.cache[key] if time.time() - entry["timestamp"] < self.ttl: return entry["value"] # Berechnen und cachen 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) ) # ... weitere Metriken 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

Tags

RAGagentsLLMorchestrationarchitectureavancélanggraphcrewai

Verwandte Artikel

Ailog Assistant

Ici pour vous aider

Salut ! Pose-moi des questions sur Ailog et comment intégrer votre RAG dans vos projets !