Einführung
Retrieval-Augmented Generation begann als eine einfache Idee: Anstatt sich ausschließlich auf das parametrische Gedächtnis eines Sprachmodells zu verlassen, relevante Dokumente aus einer externen Wissensbasis abrufen und sie in den Prompt-Kontext einbeziehen. Dieses naive RAG-Muster, bestehend aus Embed-Retrieve-Generate, funktionierte überraschend gut für einfache Fragebeantwortung über strukturierte Dokumentsammlungen. Aber als Teams RAG für komplexere Anwendungsfälle in die Produktion brachten, wurden seine Einschränkungen schmerzhaft deutlich. Abfragen, die Schlussfolgerungen über mehrere Dokumente hinweg erforderten, mehrdeutige Fragen, die Klärung benötigten, und Wissensbasen mit heterogenen Inhaltstypen legten die Brüchigkeit der Retrieve-then-Generate-Pipeline offen.
Agentisches RAG stellt einen fundamentalen Wandel in der Architektur von Retrieval-Systemen dar. Anstatt einer festen Pipeline, bei der jeder Schritt linear in den nächsten einfließt, gibt agentisches RAG dem Sprachmodell die Fähigkeit, seine Retrieval-Strategie zu planen, die Qualität abgerufener Ergebnisse zu bewerten, Abfragen umzuformulieren, wenn der anfängliche Abruf fehlschlägt, und zu entscheiden, wann es genügend Informationen hat, um eine finale Antwort zu generieren. Das Modell wird zum aktiven Teilnehmer am Retrieval-Prozess statt zum passiven Konsumenten abgerufenen Kontexts.
Dieser Leitfaden behandelt die Architekturmuster, Implementierungs-Frameworks, Evaluierungsmethoden und Produktionsüberlegungen für den Aufbau agentischer RAG-Systeme. Wir stützen uns auf reale Deployments in Unternehmens-Wissensbasen, Kundensupport-Systemen und technischen Dokumentationsplattformen, wo diese Muster im großen Maßstab validiert wurden.
Von naivem RAG zu agentischem RAG: Was sich geändert hat
Naives RAG folgt einer deterministischen Drei-Schritte-Pipeline: Die Benutzerabfrage wird embedded, eine Vektor-Ähnlichkeitssuche ruft die top-k ähnlichsten Dokumentfragmente ab, und diese Fragmente werden zu einem Prompt zusammengefügt, den das LLM zur Antwortgenerierung verwendet. Diese Pipeline hat drei fundamentale Schwächen, die agentisches RAG adressiert.
Erstens nimmt naives RAG an, dass die Benutzerabfrage bereits gut für den Abruf formuliert ist. In der Praxis sind Benutzerabfragen oft vage, vielschichtig oder verwenden Terminologie, die nicht zum Vokabular der indexierten Dokumente passt. Ein Benutzer, der nach Deployment-Fehlern fragt, könnte Dokumente über Fehlerbehandlung, Infrastrukturkonfiguration und CI/CD-Pipelines benötigen, aber eine einzelne Vektorsuche liefert möglicherweise nur eine dieser Facetten.
Zweitens hat naives RAG kein Qualitätstor. Wenn die abgerufenen Dokumente irrelevant, veraltet oder unzureichend sind, fährt die Pipeline trotzdem fort und das LLM generiert eine Antwort aus schlechtem Kontext. Es gibt keinen Mechanismus, damit das System einen Abruffehler erkennt und es erneut versucht.
Drittens behandelt naives RAG alle Abfragen identisch. Eine faktische Suchfrage, eine komplexe analytische Frage und eine Frage, die Synthese aus mehreren Quellen erfordert, durchlaufen alle dieselbe Retrieve-Generate-Pipeline. Agentisches RAG führt bedingte Logik ein, die verschiedene Retrieval- und Generierungsstrategien basierend auf Abfragemerkmalen auswählt.
Der Übergang von naivem zu agentischem RAG beinhaltet das Hinzufügen von drei Fähigkeiten: Abfrageplanung (Zerlegung komplexer Abfragen in Teilabfragen), Retrieval-Evaluierung (Bewertung, ob der abgerufene Kontext ausreichend ist) und iterative Verfeinerung (Umformulierung von Abfragen und erneuter Abruf bei unzureichenden Ergebnissen). Diese Fähigkeiten transformieren eine statische Pipeline in ein dynamisches, selbstkorrigierendes System.
Kern-Architekturmuster
Agentische RAG-Systeme werden aus einer kleinen Anzahl kombinierbarer Muster aufgebaut. Das Verständnis dieser Muster ermöglicht es Ihnen, Systeme zu entwerfen, die auf Ihren spezifischen Anwendungsfall zugeschnitten sind.
Routing
Das einfachste agentische Muster leitet Abfragen basierend auf der Abfrageklassifikation an verschiedene Retrieval-Backends weiter. Ein Router-Agent analysiert die eingehende Abfrage und leitet sie an die am besten geeignete Wissensquelle weiter:
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
router_prompt = ChatPromptTemplate.from_messages([
("system", """Classify the user query into one of these categories:
- technical_docs: Questions about API usage, configuration, or code
- policy: Questions about company policies, procedures, or compliance
- support: Questions about troubleshooting or known issues
- general: General questions that don't fit other categories
Respond with ONLY the category name."""),
("human", "{query}")
])
router_chain = router_prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()
Routing ist wertvoll, wenn Ihre Wissensbasis mehrere Domänen mit unterschiedlichen Indexierungsstrategien umfasst. Technische Dokumentation könnte mit code-bewusstem Chunking und Embeddings indexiert sein, während Richtliniendokumente semantisches Chunking mit Metadaten-Filtern verwenden.
Abfragezerlegung
Komplexe Fragen erfordern oft Informationen aus mehreren Dokumentfragmenten, die in einem einzelnen Abruf nicht zusammen erscheinen würden. Die Abfragezerlegung teilt eine komplexe Abfrage in unabhängige Teilabfragen auf, ruft für jede ab und synthetisiert die Ergebnisse:
decomposition_prompt = ChatPromptTemplate.from_messages([
("system", """Break the following complex question into 2-4 simpler
sub-questions that, when answered together, provide a complete answer
to the original question. Return as a JSON array of strings."""),
("human", "{query}")
])
# Example: "How does our API rate limiting compare to competitors
# and what are customers saying about it?"
# Decomposes to:
# ["What are our current API rate limiting policies?",
# "What rate limiting do our main competitors use?",
# "What customer feedback have we received about rate limiting?"]
Selbstkorrektur
Selbstkorrektur ist das Muster, das agentisches RAG am meisten von naivem RAG unterscheidet. Nach dem Abruf bewertet ein Bewertungsschritt, ob die abgerufenen Dokumente relevant und ausreichend sind. Wenn nicht, formuliert das System die Abfrage um und versucht es erneut:
grading_prompt = ChatPromptTemplate.from_messages([
("system", """You are a retrieval quality grader. Given a user question
and a set of retrieved documents, determine:
1. Are the documents relevant to the question? (yes/no)
2. Do the documents contain sufficient information to answer? (yes/no)
3. If insufficient, suggest a reformulated query.
Respond as JSON with keys: relevant, sufficient, reformulated_query"""),
("human", "Question: {query}\n\nDocuments: {documents}")
])
Dieses Muster erlaubt typischerweise 2-3 Abrufversuche, bevor es auf eine elegante Fehlermeldung zurückfällt. Die umformulierten Abfragen wechseln oft von semantischer Ähnlichkeitssuche zu schlüsselwortbasierter Suche, erweitern den Umfang oder zielen auf spezifische Metadatenfelder ab.
LangGraph für zustandsbehaftete Agenten-Orchestrierung
LangGraph hat sich als Standard-Framework für den Aufbau agentischer RAG-Systeme etabliert, weil es explizites Zustandsmanagement, bedingtes Routing und Zyklusunterstützung bietet, die die einfacheren kettenbasierten Abstraktionen in LangChain nicht ausdrücken können.
Ein LangGraph-basiertes agentisches RAG-System wird als Zustandsgraph definiert, bei dem Knoten Verarbeitungsschritte und Kanten Übergänge zwischen Schritten darstellen. Bedingte Kanten ermöglichen es dem Graph, basierend auf Zwischenergebnissen zu verzweigen:
from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Annotated
from operator import add
class AgentState(TypedDict):
query: str
sub_queries: List[str]
documents: List[dict]
generation: str
retry_count: int
retrieval_grade: str
def route_query(state: AgentState) -> AgentState:
"""Classify and route the incoming query."""
query = state["query"]
classification = router_chain.invoke({"query": query})
state["route"] = classification
return state
def retrieve(state: AgentState) -> AgentState:
"""Retrieve documents based on query and route."""
query = state["query"]
docs = retriever.invoke(query)
state["documents"] = docs
return state
def grade_documents(state: AgentState) -> AgentState:
"""Grade retrieved documents for relevance and sufficiency."""
grade = grading_chain.invoke({
"query": state["query"],
"documents": state["documents"]
})
state["retrieval_grade"] = grade["sufficient"]
state["retry_count"] = state.get("retry_count", 0) + 1
if not grade["sufficient"] and grade.get("reformulated_query"):
state["query"] = grade["reformulated_query"]
return state
def generate(state: AgentState) -> AgentState:
"""Generate final answer from retrieved context."""
answer = generation_chain.invoke({
"query": state["query"],
"documents": state["documents"]
})
state["generation"] = answer
return state
def should_retry(state: AgentState) -> str:
"""Decide whether to retry retrieval or proceed to generation."""
if state["retrieval_grade"] == "yes":
return "generate"
if state["retry_count"] >= 3:
return "generate" # Give up and generate with what we have
return "retrieve"
# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("route", route_query)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade", grade_documents)
workflow.add_node("generate", generate)
workflow.set_entry_point("route")
workflow.add_edge("route", "retrieve")
workflow.add_edge("retrieve", "grade")
workflow.add_conditional_edges("grade", should_retry, {
"retrieve": "retrieve",
"generate": "generate"
})
workflow.add_edge("generate", END)
app = workflow.compile()
Die Graphstruktur macht den Kontrollfluss explizit und debugbar. Sie können den Graph visualisieren, Ausführungen durch jeden Knoten verfolgen und genau verstehen, warum das System für eine gegebene Abfrage einen bestimmten Pfad gewählt hat. Diese Transparenz ist kritisch für Produktionssysteme, in denen Sie Fehler diagnostizieren und Verhalten erklären müssen.
LangGraph unterstützt auch fortgeschrittenere Muster wie parallelen Abruf über mehrere Indizes, Human-in-the-Loop-Checkpoints, bei denen der Agent für eine Benutzerbestätigung pausiert, und persistenten Zustand, der über Konversationen hinweg bestehen bleibt.
Retrieval-Strategien
Die Retrieval-Schicht in einem agentischen RAG-System ist typischerweise ausgefeilter als ein einzelner Vector-Store-Lookup. Produktionssysteme kombinieren mehrere Retrieval-Strategien.
Hybride Suche
Die hybride Suche kombiniert dichten Vektor-Retrieval mit spärlichem Keyword-Retrieval und gibt Ihnen das semantische Verständnis von Embeddings zusammen mit der Präzision von BM25-Keyword-Matching:
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
from langchain_chroma import Chroma
vectorstore = Chroma(
collection_name="documents",
embedding_function=embedding_model,
)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 10
hybrid_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.6, 0.4],
)
Re-Ranking
Nach dem initialen Abruf bewertet ein Cross-Encoder-Re-Ranker jedes Dokument gegen die Abfrage mit deutlich höherer Genauigkeit als die Bi-Encoder-Ähnlichkeit. Dies ist rechenintensiv, verbessert aber die Präzision dramatisch:
from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank
reranker = CohereRerank(model="rerank-v3.5", top_n=5)
compression_retriever = ContextualCompressionRetriever(
base_compressor=reranker,
base_retriever=hybrid_retriever,
)
Multi-Index-Retrieval
Produktions-Wissensbasen umfassen oft mehrere Indizes mit unterschiedlichen Schemas, Embedding-Modellen und Dokumenttypen. Ein agentisches System kann mehrere Indizes parallel abfragen und Ergebnisse zusammenführen:
async def multi_index_retrieve(query: str, indexes: list) -> list:
"""Retrieve from multiple indexes in parallel."""
import asyncio
async def retrieve_from_index(index, query):
return await index.aretrieve(query)
tasks = [retrieve_from_index(idx, query) for idx in indexes]
results = await asyncio.gather(*tasks)
# Flatten and deduplicate
all_docs = []
seen_ids = set()
for result_set in results:
for doc in result_set:
if doc.metadata["id"] not in seen_ids:
all_docs.append(doc)
seen_ids.add(doc.metadata["id"])
return all_docs
Evaluierung mit RAGAS und DeepEval
Die Evaluierung agentischer RAG-Systeme erfordert die Messung mehrerer Qualitätsdimensionen. RAGAS und DeepEval sind die zwei am weitesten verbreiteten Evaluierungs-Frameworks, jeweils mit unterschiedlichen Stärken.
RAGAS bietet eine Reihe referenzfreier Metriken, die die RAG-Qualität bewerten, ohne Ground-Truth-Antworten für jede Testabfrage zu benötigen:
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
)
from datasets import Dataset
eval_dataset = Dataset.from_dict({
"question": questions,
"answer": generated_answers,
"contexts": retrieved_contexts,
"ground_truth": reference_answers,
})
results = evaluate(
dataset=eval_dataset,
metrics=[
faithfulness,
answer_relevancy,
context_precision,
context_recall,
],
)
print(results)
Faithfulness misst, ob die generierte Antwort durch den abgerufenen Kontext gestützt wird und erkennt Halluzinationen. Answer Relevancy misst, ob die Antwort die Frage adressiert. Context Precision misst, ob die abgerufenen Dokumente relevant sind, und Context Recall misst, ob der Abruf alle notwendigen Informationen erfasst hat.
DeepEval erweitert die Evaluierung um zusätzliche Metriken, die besonders für agentische Systeme relevant sind:
from deepeval import evaluate
from deepeval.metrics import (
FaithfulnessMetric,
AnswerRelevancyMetric,
ContextualRelevancyMetric,
HallucinationMetric,
)
from deepeval.test_case import LLMTestCase
test_case = LLMTestCase(
input="How do I configure rate limiting?",
actual_output=generated_answer,
expected_output=reference_answer,
retrieval_context=retrieved_chunks,
)
metrics = [
FaithfulnessMetric(threshold=0.8),
AnswerRelevancyMetric(threshold=0.7),
ContextualRelevancyMetric(threshold=0.7),
HallucinationMetric(threshold=0.5),
]
evaluate(test_cases=[test_case], metrics=metrics)
Für agentisches RAG im Speziellen sollten Sie auch die Retrieval-Effizienz messen: Wie viele Retrieval-Runden benötigt das System durchschnittlich, welcher Prozentsatz der Abfragen erfordert Umformulierung, und wie verschlechtert sich die Antwortqualität über die Wiederholungsversuche hinweg. Diese Metriken sind spezifisch für die agentische Schleife und werden von Standard-RAG-Evaluierungs-Frameworks nicht abgedeckt.
Produktionsmuster
Den Übergang von agentischem RAG vom Prototyp zur Produktion erfordert die Behandlung von Zuverlässigkeits-, Sicherheits- und Betriebsanliegen, die während der Entwicklung nicht auftauchen.
Leitplanken
Jedes RAG-Produktionssystem benötigt Leitplanken, die das LLM daran hindern, schädliche, themenfremde oder faktisch nicht gestützte Antworten zu generieren:
from guardrails import Guard
from guardrails.hub import ToxicLanguage, CompetitorCheck
guard = Guard().use_many(
ToxicLanguage(on_fail="exception"),
CompetitorCheck(
competitors=["competitor_a", "competitor_b"],
on_fail="fix"
),
)
raw_response = generation_chain.invoke({"query": query, "documents": docs})
validated_response = guard.validate(raw_response)
Fallback-Ketten
Wenn die agentische Schleife nach der maximalen Anzahl von Wiederholungen keinen ausreichenden Kontext findet, benötigt das System eine elegante Degradierung anstatt eine unzuverlässige Antwort zu generieren:
def generate_with_fallback(state: AgentState) -> AgentState:
if state["retrieval_grade"] != "yes" and state["retry_count"] >= 3:
state["generation"] = (
"I wasn't able to find sufficient information in our knowledge base "
"to fully answer your question. Here's what I found:\n\n"
f"{partial_answer_from_context(state['documents'])}\n\n"
"For a complete answer, I'd recommend contacting the support team."
)
state["confidence"] = "low"
else:
state["generation"] = generation_chain.invoke({
"query": state["query"],
"documents": state["documents"]
})
state["confidence"] = "high"
return state
Human-in-the-Loop
Für Hochrisiko-Anwendungen unterstützt LangGraph Unterbrechungspunkte, an denen der Agent die Ausführung pausiert und auf menschliche Genehmigung wartet:
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
# Add interrupt before generation for sensitive queries
workflow.add_node("human_review", lambda state: state)
workflow.add_conditional_edges(
"grade",
lambda state: "human_review" if state.get("sensitive") else "generate",
{"human_review": "human_review", "generate": "generate"}
)
app = workflow.compile(
checkpointer=checkpointer,
interrupt_before=["human_review"]
)
Observability mit LangSmith und Phoenix
Observability ist nicht verhandelbar für agentische Produktionssysteme. Die nicht-deterministische Natur des LLM-gesteuerten Kontrollflusses bedeutet, dass Sie nicht jeden möglichen Ausführungspfad vorhersagen oder testen können. Sie benötigen umfassendes Tracing, um zu verstehen, was Ihr System in der Produktion tut.
LangSmith bietet End-to-End-Tracing für LangGraph-Anwendungen:
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "agentic-rag-production"
# All LangGraph invocations are automatically traced
result = app.invoke({"query": "How do I configure rate limiting?"})
Jede Trace zeigt den vollständigen Ausführungspfad durch den Graph: welche Knoten besucht wurden, was die LLM-Ein- und Ausgaben bei jedem Schritt waren, Retrieval-Ergebnisse, Bewertungsentscheidungen und Token-Zählungen. Dies ist essentiell für das Debugging von Fällen, in denen der Agent einen unerwarteten Pfad einschlägt oder eine schlechte Antwort generiert.
Arize Phoenix bietet eine Open-Source-Alternative mit Fokus auf Retrieval-Qualitätsmonitoring:
import phoenix as px
from phoenix.trace.langchain import LangChainInstrumentor
px.launch_app()
LangChainInstrumentor().instrument()
# Traces are now collected in Phoenix
result = app.invoke({"query": user_query})
Phoenix ist besonders wertvoll für die Überwachung der Retrieval-Qualität über die Zeit. Es verfolgt Embedding-Drift, Verteilungen der Retrieval-Relevanz und kann alertieren, wenn sich die Retrieval-Qualität verschlechtert, was oft darauf hinweist, dass die Wissensbasis von der Trainingsverteilung des Embedding-Modells abgedriftet ist und eine Neuindexierung benötigt.
Schlüsselmetriken für das Produktionsmonitoring:
# Custom metrics to track
metrics = {
"avg_retrieval_rounds": [],
"reformulation_rate": [],
"fallback_rate": [],
"latency_p50_ms": [],
"latency_p99_ms": [],
"token_cost_per_query": [],
"faithfulness_score": [],
}
Kosten- und Latenzoptimierung
Agentische RAG-Systeme sind von Natur aus teurer und langsamer als naives RAG, da sie mehrere LLM-Aufrufe pro Abfrage durchführen. Die Optimierung von Kosten und Latenz ohne Qualitätseinbußen erfordert sorgfältige architektonische Entscheidungen.
Verwenden Sie kleinere Modelle für Routing und Bewertung. Der Router und der Dokumentbewerter benötigen nicht die volle Reasoning-Fähigkeit eines Frontier-Modells. GPT-4o-mini oder Claude 3.5 Haiku können diese Aufgaben zu einem Bruchteil der Kosten und Latenz bewältigen:
# Use a small, fast model for routing and grading
routing_model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
grading_model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# Use a larger model only for final generation
generation_model = ChatOpenAI(model="gpt-4o", temperature=0.1)
Cachen Sie Retrieval-Ergebnisse aggressiv. Viele Benutzerabfragen sind Variationen derselben zugrundeliegenden Frage. Ein semantischer Cache, der Abfragen innerhalb eines Ähnlichkeitsschwellenwerts abgleicht, kann redundante Abrufe und LLM-Aufrufe eliminieren:
from langchain_community.cache import RedisSemanticCache
set_llm_cache(RedisSemanticCache(
redis_url="redis://localhost:6379",
embedding=embedding_model,
score_threshold=0.95,
))
Setzen Sie aggressive Timeouts für jeden Schritt, um unkontrollierte Latenz zu verhindern. Ein einzelner langsamer LLM-Aufruf sollte nicht dazu führen, dass die gesamte Anfrage in ein Timeout läuft:
from asyncio import wait_for, TimeoutError
async def retrieve_with_timeout(state, timeout_seconds=5):
try:
return await wait_for(retrieve(state), timeout=timeout_seconds)
except TimeoutError:
state["documents"] = []
state["retrieval_grade"] = "no"
return state
Praxisbeispiele und Anti-Patterns
Das häufigste Fehlermuster bei agentischen RAG-Deployments ist die unendliche Retry-Schleife. Ohne eine harte Obergrenze für Abrufversuche und klares Fallback-Verhalten kann der Agent endlos durch Umformulierungen kreisen und dabei Token und Latenz verbrennen. Setzen Sie immer explizite Retry-Limits und messen Sie Ihre Umformulierungsrate. Wenn mehr als 20% der Abfragen eine Umformulierung erfordern, liegt das Problem wahrscheinlich in Ihrer Retrieval-Schicht (schlechtes Chunking, falsches Embedding-Modell, veralteter Index) und nicht in der Agenten-Logik.
Ein weiteres häufiges Anti-Pattern ist Über-Routing. Teams bauen manchmal komplexe Routing-Graphen mit Dutzenden spezialisierter Indizes, obwohl ein einzelner gut konzipierter hybrider Retriever besser abschneiden würde. Beginnen Sie mit der einfachsten Architektur, die Ihre Anforderungen erfüllt, und fügen Sie Komplexität erst hinzu, wenn Sie Metriken haben, die zeigen, dass einfachere Ansätze unzureichend sind.
Kontextfenster-Überfüllung ist ein drittes Anti-Pattern. 20 Dokumentfragmente abzurufen und sie alle in den Prompt zu stopfen verschwendet Token und kann die Antwortqualität tatsächlich verschlechtern, weil das LLM Schwierigkeiten hat, die relevanten Informationen in einem Meer von marginal verwandtem Kontext zu identifizieren. Re-Ranking auf eine kleine Anzahl hochwertiger Fragmente (typischerweise 3-5) übertrifft in Evaluierungs-Benchmarks konsistent größere Kontextfenster.
Hüten Sie sich schließlich vor evaluierungsgetriebener Über-Optimierung. Teams, die ausschließlich für RAGAS-Scores optimieren, können Systeme bauen, die in Benchmarks gut abschneiden, aber übermäßig vorsichtige, ausweichende Antworten produzieren, die Benutzer als nutzlos empfinden. Balancieren Sie quantitative Evaluierung mit qualitativer Überprüfung tatsächlicher Benutzerkonversationen.
Die Zukunft: Multi-Agenten-RAG-Systeme
Die nächste Grenze im agentischen RAG sind Multi-Agenten-Systeme, bei denen spezialisierte Agenten bei komplexen Retrieval- und Syntheseaufgaben zusammenarbeiten. Anstatt dass ein einzelner Agent die gesamte Retrieval-Generierungs-Pipeline handhabt, setzen Sie ein Team von Agenten mit unterschiedlichen Spezialisierungen ein.
Ein Recherche-Agent übernimmt die Abfragezerlegung und Multi-Hop-Retrieval über große Dokumentsammlungen. Ein Faktencheck-Agent verifiziert generierte Behauptungen gegen Quelldokumente. Ein Synthese-Agent kombiniert Erkenntnisse aus mehreren Teilabfragen zu einer kohärenten, gut strukturierten Antwort. Ein Editor-Agent überprüft die finale Ausgabe auf Klarheit, Genauigkeit und Ton.
from langgraph.graph import StateGraph
# Multi-agent RAG architecture
multi_agent = StateGraph(MultiAgentState)
multi_agent.add_node("planner", planner_agent)
multi_agent.add_node("researcher", researcher_agent)
multi_agent.add_node("fact_checker", fact_check_agent)
multi_agent.add_node("synthesizer", synthesis_agent)
multi_agent.set_entry_point("planner")
multi_agent.add_edge("planner", "researcher")
multi_agent.add_edge("researcher", "fact_checker")
multi_agent.add_conditional_edges(
"fact_checker",
lambda s: "researcher" if s["needs_more_research"] else "synthesizer",
)
multi_agent.add_edge("synthesizer", END)
Diese Architektur ist pro Abfrage teurer, bewältigt aber komplexe Recherchefragen, mit denen Einzelagenten-Systeme Schwierigkeiten haben. Die zentrale Designherausforderung besteht darin, klare Schnittstellen zwischen Agenten zu definieren und sicherzustellen, dass der Overhead der Inter-Agenten-Kommunikation die Vorteile der Spezialisierung nicht zunichtemacht.
Das Tool-Ökosystem konvergiert darauf, Multi-Agenten-Muster nativ zu unterstützen. LangGraphs Subgraph-Komposition, LlamaIndex' Agenten-Orchestrierungsschicht und CrewAI bieten alle Primitive für den Aufbau von Multi-Agenten-RAG-Systemen. Während diese Frameworks reifen und die Kosten weiter sinken, wird Multi-Agenten-RAG für ein breiteres Spektrum von Produktionsanwendungen praktikabel. Die in diesem Leitfaden behandelten Prinzipien — explizites Zustandsmanagement, Retrieval-Qualitätsevaluierung, elegante Fallbacks und umfassende Observability — bleiben essentiell, unabhängig davon, ob Sie einen einzelnen Agenten oder ein Team spezialisierter Agenten einsetzen.