LangGraph: Komplexe RAG-Workflows

LangGraph ist ein leistungsstarkes Framework zum Aufbau von zustandsbehafteten Multi-Agent-LLM-Anwendungen. Dieser Leitfaden zeigt, wie man es nutzt, um anspruchsvolle RAG-Pipelines mit bedingtem Routing, Korrekturschleifen und Agenten-Orchestrierung zu erstellen.

Voraussetzungen: Lesen Sie die RAG-Grundlagen und unseren Leitfaden zur Orchestrierung von RAG-Agenten.

Warum LangGraph für RAG?

Einschränkungen des linearen RAG

LangGraph-Architektur

LANGGRAPH-ARCHITEKTUR

StateGraph
    |
    v
+-------+     +-------+     +-------+
| Node1 | --> | Node2 | --> | Node3 |
+-------+     +-------+     +-------+
    |             |             |
    +-------------+-------------+
                  |
                  v
            +---------+
            |  State  |
            +---------+
            (von allen Knoten gemeinsam genutzt)

Vorteile:
- Persistenter Zustand über alle Knoten hinweg
- Zyklen möglich (Korrektur, Retry)
- Routing-Bedingungen
- Parallelisierung

Grundlegende Konzepte

State, Nodes und Edges

DEVELOPERpython
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage
import operator

# 1. Definition des Zustands
class RAGState(TypedDict):
    """Gemeinsamer Zustand des RAG-Workflows."""
    # Gesprächsnachrichten
    messages: Annotated[List, operator.add]

    # Benutzeranfrage
    query: str

    # Abgerufene Dokumente
    documents: List[dict]

    # Generierte Antwort
    response: str

    # Metadaten
    retrieval_count: int
    is_relevant: bool
    needs_more_context: bool

# 2. Definition der Knoten
def retrieve_documents(state: RAGState) -> RAGState:
    """Knoten zum Abrufen von Dokumenten."""
    query = state["query"]

    # Retrieval-Simulation
    documents = retriever.search(query, top_k=5)

    return {
        "documents": documents,
        "retrieval_count": state.get("retrieval_count", 0) + 1
    }

def grade_documents(state: RAGState) -> RAGState:
    """Knoten zur Bewertung der Dokumente."""
    documents = state["documents"]
    query = state["query"]

    # Relevanz bewerten
    relevant_docs = []
    for doc in documents:
        score = evaluate_relevance(doc, query)
        if score > 0.7:
            relevant_docs.append(doc)

    is_relevant = len(relevant_docs) >= 2

    return {
        "documents": relevant_docs,
        "is_relevant": is_relevant
    }

def generate_response(state: RAGState) -> RAGState:
    """Knoten zur Antwortgenerierung."""
    query = state["query"]
    documents = state["documents"]

    # Kontext aufbauen
    context = "\n\n".join([doc["content"] for doc in documents])

    # Antwort generieren
    prompt = f"""Context:\n{context}\n\nQuestion: {query}\n\nAnswer:"""
    response = llm.invoke(prompt)

    return {
        "response": response.content,
        "messages": [AIMessage(content=response.content)]
    }

def check_hallucination(state: RAGState) -> RAGState:
    """Prüft, ob die Antwort Halluzinationen enthält."""
    response = state["response"]
    documents = state["documents"]

    # Prüfen, ob die Antwort auf den Dokumenten basiert
    is_grounded = verify_grounding(response, documents)

    return {
        "needs_more_context": not is_grounded
    }

# 3. Aufbau des Graphen
def build_rag_graph():
    """Baut den RAG-Graphen mit LangGraph auf."""
    workflow = StateGraph(RAGState)

    # Knoten hinzufügen
    workflow.add_node("retrieve", retrieve_documents)
    workflow.add_node("grade", grade_documents)
    workflow.add_node("generate", generate_response)
    workflow.add_node("check", check_hallucination)

    # Edges definieren
    workflow.set_entry_point("retrieve")
    workflow.add_edge("retrieve", "grade")

    # Bedingter Edge nach der Bewertung
    workflow.add_conditional_edges(
        "grade",
        lambda state: "generate" if state["is_relevant"] else "retrieve",
        {
            "generate": "generate",
            "retrieve": "retrieve"  # Retry-Schleife
        }
    )

    workflow.add_edge("generate", "check")

    # Bedingter Edge nach der Prüfung
    workflow.add_conditional_edges(
        "check",
        lambda state: END if not state["needs_more_context"] else "retrieve",
        {
            END: END,
            "retrieve": "retrieve"  # Retry bei Halluzination
        }
    )

    return workflow.compile()

RAG-Workflow-Patterns

Pattern 1: Self-RAG (Automatische Korrektur)

DEVELOPERpython
from langgraph.graph import StateGraph, END
from typing import Literal

class SelfRAGState(TypedDict):
    query: str
    documents: List[dict]
    response: str
    reflection: str
    iteration: int
    max_iterations: int

def reflect_on_response(state: SelfRAGState) -> SelfRAGState:
    """Reflexion über die Qualität der Antwort."""
    response = state["response"]
    query = state["query"]

    reflection_prompt = f"""
    Bewerte diese Antwort auf die Frage.

    Frage: {query}
    Antwort: {response}

    Kriterien:
    1. Ist die Antwort vollständig?
    2. Ist sie faktisch korrekt und überprüfbar?
    3. Beantwortet sie die Frage direkt?

    Falls Verbesserungen nötig sind, erkläre welche.
    Andernfalls antworte "SATISFAISANT".
    """

    reflection = llm.invoke(reflection_prompt)

    return {
        "reflection": reflection.content,
        "iteration": state["iteration"] + 1
    }

def should_continue(state: SelfRAGState) -> Literal["improve", "end"]:
    """Entscheidet, ob die Verbesserung fortgesetzt wird."""
    if state["iteration"] >= state["max_iterations"]:
        return "end"
    if "SATISFAISANT" in state["reflection"]:
        return "end"
    return "improve"

def improve_response(state: SelfRAGState) -> SelfRAGState:
    """Verbessert die Antwort basierend auf der Reflexion."""
    improve_prompt = f"""
    Verbessere diese Antwort basierend auf dem Feedback.

    Ursprüngliche Frage: {state['query']}
    Aktuelle Antwort: {state['response']}
    Feedback: {state['reflection']}

    Neue, verbesserte Antwort:
    """

    improved = llm.invoke(improve_prompt)

    return {"response": improved.content}

def build_self_rag():
    workflow = StateGraph(SelfRAGState)

    workflow.add_node("retrieve", retrieve_documents)
    workflow.add_node("generate", generate_response)
    workflow.add_node("reflect", reflect_on_response)
    workflow.add_node("improve", improve_response)

    workflow.set_entry_point("retrieve")
    workflow.add_edge("retrieve", "generate")
    workflow.add_edge("generate", "reflect")

    workflow.add_conditional_edges(
        "reflect",
        should_continue,
        {"improve": "improve", "end": END}
    )

    workflow.add_edge("improve", "reflect")  # Korrekturschleife

    return workflow.compile()

Pattern 2: Adaptive RAG (Intelligentes Routing)

DEVELOPERpython
class AdaptiveRAGState(TypedDict):
    query: str
    query_type: str  # "simple", "complex", "comparison"
    documents: List[dict]
    sub_queries: List[str]
    sub_answers: List[str]
    final_response: str

def classify_query(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """Klassifiziert die Komplexität der Anfrage."""
    query = state["query"]

    classification_prompt = f"""
    Klassifiziere diese Frage:
    "{query}"

    Mögliche Typen:
    - simple: Direkte Faktenfrage
    - complex: Erfordert mehrere Denkschritte
    - comparison: Vergleicht mehrere Elemente

    Antworte nur mit dem Typ.
    """

    query_type = llm.invoke(classification_prompt).content.strip().lower()

    return {"query_type": query_type}

def decompose_query(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """Zerlegt eine komplexe Anfrage in Teilanfragen."""
    query = state["query"]

    decompose_prompt = f"""
    Zerlege diese komplexe Frage in Teilfragen:
    "{query}"

    Liste jede Teilfrage in einer eigenen Zeile.
    """

    result = llm.invoke(decompose_prompt)
    sub_queries = [q.strip() for q in result.content.split("\n") if q.strip()]

    return {"sub_queries": sub_queries}

def process_sub_queries(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """Verarbeitet jede Teilanfrage einzeln."""
    sub_answers = []

    for sub_query in state["sub_queries"]:
        # Retrieval für jede Teilanfrage
        docs = retriever.search(sub_query, top_k=3)
        context = "\n".join([d["content"] for d in docs])

        answer = llm.invoke(f"Context: {context}\n\nQuestion: {sub_query}")
        sub_answers.append(answer.content)

    return {"sub_answers": sub_answers}

def synthesize_response(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """Fasst die Teilantworten zusammen."""
    synthesis_prompt = f"""
    Ursprüngliche Frage: {state['query']}

    Antworten auf die Teilfragen:
    {chr(10).join([f'- {a}' for a in state['sub_answers']])}

    Fasse eine vollständige und kohärente Antwort zusammen.
    """

    final = llm.invoke(synthesis_prompt)

    return {"final_response": final.content}

def route_by_complexity(state: AdaptiveRAGState) -> str:
    """Routet je nach Komplexität der Anfrage."""
    query_type = state["query_type"]

    if query_type == "simple":
        return "simple_path"
    elif query_type == "complex":
        return "complex_path"
    else:
        return "comparison_path"

def build_adaptive_rag():
    workflow = StateGraph(AdaptiveRAGState)

    # Knoten
    workflow.add_node("classify", classify_query)
    workflow.add_node("simple_retrieve", retrieve_documents)
    workflow.add_node("simple_generate", generate_response)
    workflow.add_node("decompose", decompose_query)
    workflow.add_node("process_subs", process_sub_queries)
    workflow.add_node("synthesize", synthesize_response)

    # Edges
    workflow.set_entry_point("classify")

    workflow.add_conditional_edges(
        "classify",
        route_by_complexity,
        {
            "simple_path": "simple_retrieve",
            "complex_path": "decompose",
            "comparison_path": "decompose"
        }
    )

    workflow.add_edge("simple_retrieve", "simple_generate")
    workflow.add_edge("simple_generate", END)

    workflow.add_edge("decompose", "process_subs")
    workflow.add_edge("process_subs", "synthesize")
    workflow.add_edge("synthesize", END)

    return workflow.compile()

Pattern 3: Multi-Agent RAG

DEVELOPERpython
class MultiAgentState(TypedDict):
    query: str
    documents: List[dict]
    researcher_notes: str
    writer_draft: str
    reviewer_feedback: str
    final_response: str
    revision_count: int

def researcher_agent(state: MultiAgentState) -> MultiAgentState:
    """Rechercheur-Agent: analysiert die Dokumente."""
    docs = state["documents"]

    researcher_prompt = """
    Du bist ein Rechercheur. Analysiere diese Dokumente und extrahiere:
    1. Die wichtigsten Fakten
    2. Die relevanten Quellen
    3. Konsens- und Streitpunkte

    Dokumente:
    {docs}

    Recherchenotizen:
    """

    notes = llm.invoke(researcher_prompt.format(
        docs="\n".join([d["content"] for d in docs])
    ))

    return {"researcher_notes": notes.content}

def writer_agent(state: MultiAgentState) -> MultiAgentState:
    """Autor-Agent: verfasst eine Antwort."""
    writer_prompt = f"""
    Du bist ein Autor. Nutze diese Recherchenotizen, um zu antworten:

    Frage: {state['query']}

    Recherchenotizen:
    {state['researcher_notes']}

    Verfasse eine klare und strukturierte Antwort.
    """

    draft = llm.invoke(writer_prompt)

    return {"writer_draft": draft.content}

def reviewer_agent(state: MultiAgentState) -> MultiAgentState:
    """Reviewer-Agent: prüft die Qualität."""
    reviewer_prompt = f"""
    Du bist ein anspruchsvoller Reviewer. Bewerte diese Antwort:

    Frage: {state['query']}
    Antwort: {state['writer_draft']}

    Kriterien:
    1. Faktische Genauigkeit
    2. Klarheit
    3. Vollständigkeit
    4. Stil

    Falls akzeptabel, antworte "APPROUVÉ".
    Andernfalls gib Verbesserungsvorschläge.
    """

    feedback = llm.invoke(reviewer_prompt)

    return {
        "reviewer_feedback": feedback.content,
        "revision_count": state.get("revision_count", 0) + 1
    }

def revise_draft(state: MultiAgentState) -> MultiAgentState:
    """Überarbeitet den Entwurf basierend auf dem Feedback."""
    revise_prompt = f"""
    Überarbeite diese Antwort gemäß dem Feedback:

    Aktuelle Antwort: {state['writer_draft']}
    Feedback: {state['reviewer_feedback']}

    Überarbeitete Antwort:
    """

    revised = llm.invoke(revise_prompt)

    return {"writer_draft": revised.content}

def finalize_response(state: MultiAgentState) -> MultiAgentState:
    """Finalisiert die Antwort."""
    return {"final_response": state["writer_draft"]}

def should_revise(state: MultiAgentState) -> str:
    """Entscheidet, ob eine Überarbeitung nötig ist."""
    if "APPROUVÉ" in state["reviewer_feedback"]:
        return "finalize"
    if state["revision_count"] >= 3:
        return "finalize"
    return "revise"

def build_multi_agent_rag():
    workflow = StateGraph(MultiAgentState)

    workflow.add_node("retrieve", retrieve_documents)
    workflow.add_node("research", researcher_agent)
    workflow.add_node("write", writer_agent)
    workflow.add_node("review", reviewer_agent)
    workflow.add_node("revise", revise_draft)
    workflow.add_node("finalize", finalize_response)

    workflow.set_entry_point("retrieve")
    workflow.add_edge("retrieve", "research")
    workflow.add_edge("research", "write")
    workflow.add_edge("write", "review")

    workflow.add_conditional_edges(
        "review",
        should_revise,
        {"revise": "revise", "finalize": "finalize"}
    )

    workflow.add_edge("revise", "review")  # Überarbeitungsschleife
    workflow.add_edge("finalize", END)

    return workflow.compile()

Ausführung und Monitoring

Streaming der Ergebnisse

DEVELOPERpython
async def stream_rag_response(
    graph,
    query: str
):
    """Streamt die Ergebnisse des RAG-Workflows."""
    initial_state = {
        "query": query,
        "documents": [],
        "response": "",
        "messages": [HumanMessage(content=query)],
        "retrieval_count": 0,
        "is_relevant": False,
        "needs_more_context": False
    }

    async for event in graph.astream(initial_state):
        # event enthält den Namen des Knotens und den aktualisierten Zustand
        node_name = list(event.keys())[0]
        state_update = event[node_name]

        yield {
            "node": node_name,
            "update": state_update
        }

# Verwendung
async def main():
    graph = build_rag_graph()

    async for update in stream_rag_response(graph, "Was ist RAG?"):
        print(f"[{update['node']}] {update['update']}")

Zustandspersistenz

DEVELOPERpython
from langgraph.checkpoint.sqlite import SqliteSaver

def build_persistent_rag():
    """RAG mit Zustandspersistenz für mehrstufige Konversationen."""

    workflow = StateGraph(RAGState)
    # ... Workflow-Konfiguration ...

    # Checkpointer für die Persistenz hinzufügen
    memory = SqliteSaver.from_conn_string(":memory:")

    return workflow.compile(checkpointer=memory)

# Verwendung mit thread_id für Konversationen
graph = build_persistent_rag()

# Erste Nachricht
config = {"configurable": {"thread_id": "user_123"}}
result1 = graph.invoke({"query": "Erkläre RAG"}, config)

# Folgenachricht (gleicher Thread)
result2 = graph.invoke({"query": "Und Embeddings?"}, config)

Best Practices

Fehlerbehandlung

DEVELOPERpython
from langgraph.graph import StateGraph

class SafeRAGState(TypedDict):
    query: str
    documents: List[dict]
    response: str
    error: str
    retry_count: int

def safe_retrieve(state: SafeRAGState) -> SafeRAGState:
    """Retrieval mit Fehlerbehandlung."""
    try:
        docs = retriever.search(state["query"])
        return {"documents": docs, "error": ""}
    except Exception as e:
        return {
            "error": str(e),
            "retry_count": state.get("retry_count", 0) + 1
        }

def handle_error(state: SafeRAGState) -> SafeRAGState:
    """Behandelt Fehler auf saubere Weise."""
    return {
        "response": f"Entschuldigung, es ist ein Fehler aufgetreten: {state['error']}"
    }

def route_on_error(state: SafeRAGState) -> str:
    if state.get("error"):
        if state.get("retry_count", 0) < 3:
            return "retry"
        return "error_handler"
    return "continue"

Implementierungs-Checkliste

• Zustand mit allen notwendigen Feldern klar definiert
• Atomare und testbare Knoten
• Klare Routing-Bedingungen
• Schutz vor Endlosschleifen (max_iterations)
• Fehlerbehandlung in jedem Knoten
• Streaming für Echtzeit-UX konfiguriert
• Persistenz für mehrstufige Konversationen
• Monitoring und Logging der Übergänge

Fazit

LangGraph ermöglicht den Aufbau anspruchsvoller RAG-Pipelines mit automatischer Korrektur, intelligentem Routing und Multi-Agent-Zusammenarbeit. Der Schlüssel liegt in der sorgfältigen Definition des gemeinsamen Zustands und der Übergangsbedingungen.

Weiterführende Ressourcen

• AutoGen: Multi-Agent-Systeme
• CrewAI: Agenten-Teams
• Function Calling RAG

Brauchen Sie orchestriertes RAG? Ailog bietet RAG-Lösungen mit intelligenten Workflows und automatischer Korrektur. Moderne und skalierbare Architektur.