LangGraph: KI-Workflows in Python mit Agenten-Templates optimieren

LangGraph v38 bietet Python-Entwicklern ein mächtiges Framework, um KI-Agenten mit strukturierten Workflows zu bauen. Doch wie setzt man Zustandsmanagement, RAG-Integration oder menschliche Eingriffe effizient um? Die neuen Templates zeigen konkrete Lösungen für typische Herausforderungen in der KI-Entwicklung.

Warum LangGraph für KI-Agenten?

LangGraph unterscheidet sich von klassischen KI-Bibliotheken durch sein zustandsbasiertes System, das komplexe Abläufe in wiederverwendbare Komponenten zerlegt. Die vier Säulen – Nodes, Edges, State und Checkpointing – ermöglichen es, Workflows dynamisch zu gestalten und Fehler leicht zu beheben. Entwickler können so Agenten erstellen, die nicht nur reagieren, sondern auch Kontext verstehen und Entscheidungen treffen.

Die Kernkomponenten im Überblick

• Nodes: Jeder Schritt im Workflow wird als Node definiert. Beispiel: Datenabruf, Textgenerierung oder Validierung.
• Edges: Definieren die Verbindungen zwischen Nodes und steuern die Ausführungsreihenfolge.
• State: Verwaltet den gesamten Kontext eines Agenten – von Eingaben bis zu Zwischenergebnissen.
• Checkpointing: Speichert den Zustand zu jedem Zeitpunkt, was Debugging und Wiederaufnahme erleichtert.

Ein einfaches Beispiel für die Initialisierung eines Graphen:

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class GraphState(TypedDict):
    messages: Annotated[list, operator.add]
    context: str

workflow = StateGraph(GraphState)

Template 1: RAG-Agent für präzise Antworten

Retrieval-Augmented Generation (RAG) kombiniert Wissensdatenbanken mit KI-Modellen, um fundierte Antworten zu liefern. Das Template zeigt, wie man einen Agenten baut, der Suchergebnisse analysiert, Kontext extrahiert und Antworten validiert.

Schritt-für-Schritt-Aufbau

1. Datenabruf: Ein Fragen-String wird an einen Retriever übergeben, der passende Dokumente findet.
2. Generierung: Die gefundenen Inhalte werden in einen Prompt integriert und an das KI-Modell gesendet.
3. Validierung: Die Antwort wird auf Plausibilität und Länge geprüft.

Ein Ausschnitt aus dem Code:

from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

class RAGAgent:
    def __init__(self, llm):
        self.llm = llm
        self.retriever = self._setup_retriever()

    def retrieve(self, state):
        question = state["messages"][-1].content
        docs = self.retriever.invoke(question)
        context = "\n\n".join([doc.page_content for doc in docs])
        return {"context": context}

    def generate(self, state):
        chain = self.prompt | self.llm | StrOutputParser()
        result = chain.invoke({
            "question": state["messages"][-1].content,
            "context": state["context"]
        })
        return {"messages": [result]}

Template 2: Multi-Tool-Agent für komplexe Aufgaben

Wenn ein Agent mehrere Tools nutzen muss – etwa Wetterdaten abrufen und Berechnungen durchführen – hilft dieses Template. Es strukturiert den Workflow in vier Phasen: Planung, Ausführung, Beobachtung und Entscheidung.

Beispiel: Wettervorhersage und Rechner

• Planung: Der Agent erstellt eine Aufgabenliste mit Prioritäten.
• Ausführung: Tools werden nacheinander oder parallel genutzt.
• Beobachtung: Die Ergebnisse werden analysiert.
• Entscheidung: Basierend auf den Daten wird eine finale Antwort generiert.

Beispiel-Tools im Code:

from langchain.tools import Tool

class MultiToolAgent:
    def __init__(self):
        self.tools = [
            Tool(
                name="wetter",
                func=lambda location: f"{location} hat sonniges Wetter.",
                description="Abfrage von Wetterinformationen"
            ),
            Tool(
                name="rechner",
                func=lambda ausdruck: f"Ergebnis: {eval(ausdruck)}",
                description="Mathematische Berechnungen"
            )
        ]

Template 3: Workflows mit menschlicher Kontrolle

In sensiblen Bereichen wie Medizin oder Recht ist menschliche Aufsicht entscheidend. Dieses Template pausiert den Workflow, fordert Feedback an und setzt die Verarbeitung erst nach Bestätigung fort.

Anwendungsfälle

• Automatisierte Dokumentenprüfung: Ein Agent markiert potenzielle Fehler und wartet auf manuelle Freigabe.
• Kritische Entscheidungen: Vor wichtigen Schritten wird ein Experte eingebunden.

Die Implementierung erfordert eine Schnittstelle für Benutzerfeedback:

class HumanInLoopAgent:
    def __init__(self):
        self.user_feedback = None

    def pause_for_review(self, state):
        return {"status": "wartend_auf_pruefung"}

    def review(self, state):
        if self.user_feedback:
            return {"messages": [self.user_feedback]}
        return {"messages": ["Bitte Feedback eingeben."]}

Template 4: Parallele Verarbeitung für Skalierung

Große Datensätze erfordern effiziente Verarbeitung. Dieses Template nutzt Thread-Pooling, um Aufgaben gleichzeitig abzuarbeiten und Ergebnisse zu aggregieren.

Vorteile

• Schnellere Durchlaufzeiten: Aufgaben werden gebündelt und parallelisiert.
• Reduzierte Serverlast: Ressourcen werden optimal genutzt.

Ein Beispiel für die Aufteilung und Verarbeitung:

from concurrent.futures import ThreadPoolExecutor

class ParallelAgent:
    def __init__(self, max_workers=4):
        self.max_workers = max_workers

    def parallel_process(self, state):
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = [
                executor.submit(self.process_chunk, chunk)
                for chunk in state["chunks"]
            ]
            results = [future.result() for future in futures]
        return {"processed_results": results}

Zustandsmanagement und Fehlerbehandlung

LangGraphs Checkpointing-System ermöglicht es, den Zustand eines Agenten zu einem beliebigen Zeitpunkt zu speichern. Das ist besonders nützlich für:

• Debugging: Fehler können nachvollzogen und behoben werden.
• Wiederherstellung: Bei Abstürzen wird der letzte gültige Zustand wiederhergestellt.

Ein Beispiel für die Integration von Speichercheckpoints:

from langgraph.checkpoint.memory import MemorySaver

class StateManager:
    def __init__(self):
        self.checkpointer = MemorySaver()

    def get_state(self, thread_id):
        return self.checkpointer.get(thread_id)

Streaming für Echtzeit-Interaktion

Echtzeit-Anwendungen wie Chatbots profitieren von Streaming, bei dem Antworten tokenweise ausgegeben werden. LangGraph unterstützt dies durch Callback-Mechanismen, die den Fortschritt sichtbar machen.

Implementierung

from langchain_core.callbacks import BaseCallbackHandler

class StreamingHandler(BaseCallbackHandler):
    def on_llm_new_token(self, token, **kwargs):
        print(token, end="", flush=True)

agent = StreamingAgent()
response = agent.stream_response("Erkläre Quantencomputing")

Fazit: LangGraph für zukunftssichere KI-Workflows

Die vorgestellten Templates zeigen, wie LangGraph Entwicklern hilft, KI-Agenten mit Python effizient zu bauen. Von einfachen RAG-Systemen bis hin zu komplexen Multi-Tool-Workflows – die Kombination aus Zustandsmanagement, Parallelverarbeitung und menschlicher Kontrolle macht LangGraph zu einem starken Werkzeug für die nächste Generation von KI-Anwendungen. Wer diese Prinzipien verinnerlicht, kann skalierbare und wartbare Lösungen schaffen, die sich an veränderte Anforderungen anpassen lassen.