Agentbasiertes RAG in LangGraph: Bewertungssysteme, bedingte Routenplanung und Websuche als Fallback

Ein semantisches RAG-Pipeline, das aus den Kernkomponenten von LangChain, einem FastAPI-Backend sowie einer Streamlit-Chatbot-Oberfläche besteht, verarbeitet PDF-Dateien, wandelt die Dokumente in Embeddings um und beantwortet Fragen über eine Chat-Schnittstelle. Es weist jedoch eine grundlegende Einschränkung auf: LangChains Primitive erzeugen lineare Abläufe – eine direkte Verbindung vom Prompt über das LLM bis zum Ausgabeparser. Alles, was bedingte Verzweigungen, Schleifen, dynamische Routing-Strategien oder Entscheidungen durch das LLM erfordert, fällt außerhalb dieses Formats, und einfache if/else-Anweisungen ermöglichen lediglich einen flachen, im Voraus definierten Ablauf.

LangGraph löst dieses Problem. Das Ergebnis ist ein agiler Workflow mit folgenden Eigenschaften:

• bedingte Verzweigungen und Schleifen, die direkt in das Pipeline integriert sind
• ein LLM, das zwischen Web-Suche und RAG-Auswertung wählt, ohne fest codierte Regeln
• Übergänge, die über LangGraph-Node und -Kanten gesteuert werden

Basierte auf LLMs entwickelte Bewertungssysteme, die den abgerufenen Kontext sowie generierte Antworten bewerten

Eine Routing-Logik, die auf diese Bewertungen reagiert

Der Zustand, der über Übergänge durch einen StateGraph übertragen wird

Das fertige Graphenmodell, das in einen FastAPI-Backend eingebunden wird, um einen produktionsreifen Chatbot zu betreiben

Am Ende erhält man ein agiles RAG-System, das aus dem Vektorlagern oder vom Web antworten kann.

Begriffe aus LangGraph

Vier Konzepte bilden das Grundvokabular. Der StateGraph steuert die Ausführung. Ein Zustand ist ein typisiertes Dictionary, das die Daten des Pipelines speichert und aktualisiert. Knoten sind Python-Funktionen, die den Zustand ändern und als Recheneinheiten des Graphen fungieren. Kanten stellen die Übergänge zwischen Knoten dar, wobei bedingte Kanten sowohl Verzweigungen als auch Schleifen ermöglichen.

Konfiguration

Platzieren Sie die API-Schlüssel in einer .env-Datei im Projektverzeichnis:

OPENAI_API_KEY="<Your OpenAI API Key>"
VECTOR_DB=pinecone
Hugging_Face_TOKEN=<your huggingface token>
PINECONE_API_KEY=<your Pinecone API Key>
PINECONE_ENVIRONMENT=gcp-starter
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"
TAVILY_API_KEY=<Your TAVILY API Key>

Dann lesen Sie sie in einer Pydantic BaseSettings-Klasse ein:

class BaseConfig(BaseSettings):
    OPENAI_API_KEY: Optional[str]
    PINECONE_API_KEY: Optional[str]
    TAVILY_API_KEY: Optional[str]

    """Loads the dotenv file. Including this is necessary to get
    pydantic to load a .env file."""
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

Projektstruktur

Die Ordnerstruktur sieht so aus:

project/
├── rag_process/
│   ├── __init__.py
│   ├── extractor.py
|   └── service.py
│
├── graphs
│   │── chains.py       # llm chains
│   ├── nodes.py        # graph nodes. Defined as Python functions
│   ├── graph.py        # graph workflow
│   ├── consts.py       # define node names as constants
│   └── state.py        # typed dictionary to store graph states
|── schemas.py          # pydantic data models
|── upload.py           # python functions to upload files asynchronously
├── main.py             # fastAPI app
├── config.py           # pydantic BaseConfig
│── dependencies.py     # dependency inject to fastAPI endpoint function
├── test_chains.py      # pytests
├── .env                # environment variable definition
├── client.py           # streamlit client
├── .gitignore
└── requirements.txt     # package requirements

Die Abrufketten, der FastAPI-Service und der Streamlit-Klient stammen aus dem früheren Pipeline. Neu sind hier das Graph, seine Zustandsklasse sowie die Bewertungsfunktionen, die die abgerufenen Dokumente auf Relevanz überprüfen, Halluzinationen erkennen und beurteilen, ob eine Antwort zur Frage passt.

Die LLM-Ketten

Die Ketten befinden sich in graphs/chains.py und verwenden die LangChain Expression Language (LCEL). Es sind sieben definiert:

• retrieve_documents holt die top-K ähnlichsten Dokumente von Pinecone
• generate_answer erstellt eine Antwort auf die Frage zusammen mit Kontext
• retrival_grader entscheidet, ob die abgerufenen Dokumente relevant sind
• web_search führt eine Internetsuche durch und liefert Kontext
• answer_grader prüft, ob die Antwort die Frage beantwortet

hallucination_grader überprüft, ob die Antwort im Kontext verankert ist

select_routes sendet Fragen zu Agenten, Prompt-Engineering und feindlichen Angriffen an RAG und alles Weitere an die Websuche

Unit-Tests für Ketten

In agierenden Systemen sind Tests wichtiger als üblich, da die Ausgaben von LLMs nicht deterministisch sind und sich ändern können, wenn Modelle aktualisiert werden, Hersteller sie anpassen oder Embeddings sich verändern. Die Tests in test_chains.py verwenden pytest zusammen mit pytest.mark.asyncio für asynchrone Ketten. Ein positiver Test für die RAG-Aufrufung sieht so aus:

@pytest.mark.asyncio
async def test_rag_retrieval_yes() -> None:
    prompt = "what is prompt engineer ?"
    searched_documents = await retrieve_documents(prompt)
    searched_document = searched_documents[0]
    # logger.info(f"searched_content: {searched_document.page_content}")
    retrieval_grade = await retrieval_grader(searched_document, prompt)
    assert retrieval_grade

Dieser Test befragt zum Prompt-Engineering und gibt das erste abgerufene Dokument an retrieval_grader weiter. Da dieses Thema bereits in den Dokumenten von Pinecone behandelt wird, sollte der Grader True zurückgeben. Der Test schützt vor Fehlern bei Änderungen der Embeddings oder des Modellverhaltens, erkennt frühzeitig Regressionen durch Modul aktualisierungen und überprüft die Routing-Logik zwischen RAG und Web-Suche. Ein entsprechender negativer Test bestätigt, dass der Grader False zurückgibt und der Ablauf zur Web-Suche weitergeleitet wird.

Zustand und Knoten

Der Zustand ist ein typisierter Wörterbuch, der in graphs/state.py gespeichert wird:

class GraphState(TypedDict):

    question: str
    generation: str
    web_search: bool
    documents: list[str]

Node lesen und aktualisieren diesen Zustand. Sie sind in graphs/nodes.py definiert und stellen gewöhnliche Python-Funktionen dar, die den Zustand lesen, LLM-Ketten aufrufen und aktualisierte Felder zurückgeben. Ein Node, der in einer bedingten Kante verwendet wird, gibt den Namen des nächsten zu ausführenden Nodes an. Es gibt insgesamt sieben.

• generate ruft generate_answer auf und gibt die Antwort zurück.
• retrieve ruft retrieve_documents auf und gibt die Dokumente zurück.
• search_web ruft web_search auf und fügt die Ergebnisse den Dokumenten hinzu.
• grade_documents behält die relevanten Dokumente bei und setzt bei Bedarf web_search=True.
• grade_generation führt die Bewertungsfunktionen für Halluzinationen und Antworten aus und gibt „nützlich“, „nicht nützlich“ oder „nicht unterstützt“ zurück.

decide_to_generate wählt je nach web_search entweder WEBSEARCH oder GENERATE aus.

route_question sendet agentbezogene Fragen an RAG und andere an die Websuche.

Aufbau des StateGraph

Der Graph selbst wird in graphs/graph.py erstellt, und der fertige Workflow ähnelt einem verzweigten Ablauf zwischen Abruf, Bewertung, Erstellung und Suche. Sein Aufbau erfordert mehrere Schritte.

Zuerst erhält jeder Knoten einen Zeichenketten-Identifikator. Das Speichern dieser Namen als Konstanten in graphs/consts.py macht den Graphen leichter lesbar und verhindert Tippfehler:

RETRIEVE = "retrieve"
GRADE_DOCUMENTS = "grade_documents"
GENERATE = "generate"
WEBSEARCH = "websearch"

Dann erstellen Sie den StateGraph und registrieren Sie jeden Knoten mit seiner Funktion:

workflow = StateGraph(GraphState)

workflow.add_node(RETRIEVE, retrieve)
workflow.add_node(GRADE_DOCUMENTS, grade_documents)
workflow.add_node(GENERATE, generate)
workflow.add_node(WEBSEARCH, search_web)

Jeder Aufruf von add_node verknüpft einen Namen mit einer Funktion, und diese Trennung zwischen Namen und Verhalten macht die Workflow-Verwaltung, Visualisierung und Erweiterung einfacher.

Drittens sollte ein bedingter Einstiegspunkt festgelegt werden. Anstatt einen Startknoten hardzukodieren, ermöglicht LangGraph es einer Funktion, die Frage zu prüfen und den ersten Weg auszuwählen. Hier entscheidet route_question, ob mit einer Websuche oder mit RAG-Auswertung begonnen werden soll:

workflow.set_conditional_entry_point(
    route_question,
    {
        WEBSEARCH: WEBSEARCH,
        RETRIEVE: RETRIEVE,
    }
)

Diese Funktion wird zuerst ausgeführt und gibt je nach Thema entweder "WEBSEARCH" oder "RETRIEVE" zurück, wobei LangGraph den Rückgabewert verwendet, um den Startknoten auszuwählen. Fragen im Zusammenhang mit Agenten werden an RAG weitergeleitet, während allgemeines Wissen durch eine Websuche ermittelt wird.

Viertens werden die Knoten mit Kanten verbunden. Statische Kanten folgen immer einem bestimmten Pfad, während bedingte Kanten den nächsten Knoten auf der Grundlage einer Entscheidung des LLM auswählen:

workflow.add_edge(RETRIEVE, GRADE_DOCUMENTS)
workflow.add_conditional_edges(
    GRADE_DOCUMENTS,
    decide_to_generate,
    {
        WEBSEARCH: WEBSEARCH,
        GENERATE: GENERATE,
    },
)

workflow.add_conditional_edges(
    GENERATE,
    grade_generation,
    {
        "not supported": GENERATE,
        "useful": END,
        "not useful": WEBSEARCH,
    }
)

workflow.add_edge(WEBSEARCH, GENERATE)
workflow.add_edge(GENERATE, END)

Zusammen bilden diese Kanten einen dynamischen Workflow, bei dem das LLM den Anfangspfad wählt, die Relevanz des Dokuments entscheidet, ob RAG oder Websuche verwendet wird, und die Qualität der Antwort bestimmt, ob weiter verfeinert, auf ein Alternativverfahren umgestellt oder abgeschlossen wird. Diese Struktur verwandelt einen einfachen RAG-Pipeline in ein stabileres agierendes System.

Zum Schluss wird das Graph zusammengefasst und gegebenenfalls dargestellt:

app = workflow.compile()

app.get_graph().draw_mermaid_png(output_file_path="graphs.png")

Ausliefern über FastAPI

In dependencies.py importiert eine Funktion das kompilierte Graph und ruft es asynchron auf:

from graphs.graph import app


async def get_generation(body: RAGRequest=Body(...)) -> dict:
    try:
        generation = await app.ainvoke(body.dict())

        return {
            "answer": generation.get("generation", ""),
            "web_search": generation.get("web_search", False)
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

FastAPI injiziert es als Abhängigkeit, und es gibt die generierte Antwort zusammen mit einem Flag zurück, das anzeigt, ob sie aus RAG oder einer Websuche stammt.

Ausprobieren

Die Streamlit-Oberfläche aus dem vorherigen Pipeline kann unverändert wiederverwendet werden. Die Anfrage, wie man richtig Millionär wird, leitet zur Websuche weiter, während die Frage nach Prompt Engineering den Weg über das semantische RAG-Verfahren nimmt. Das zeigt eine dynamische Routing-Logik zusammen mit einer klaren Herkunftsangabe für jede Antwort.

Zusammenfassend lässt sich sagen: Der LangGraph StateGraph verwaltet nun den Zustand, der Workflow weist dynamische Übergänge mit Schleifen und Verzweigungen über bedingte Kanten auf, die LLM-Grader prüfen die Relevanz der Dokumente, die Erkennung von Halluzinationen sowie die Qualität der Antworten, und die API informiert den Benutzer darüber, ob eine Antwort aus RAG oder aus dem Internet stammt. Der nächste Schritt in dieser Arbeit ist der Einsatz von Tools sowie das ReAct-Framework, um noch agileres Verhalten zu erreichen.