Erstellen Sie Ihre ersten LangGraph-Arbeitsabläufe, bevor Sie mit mehreren Agentensystemen arbeiten.

Was ist LangGraph?

Normale Python-Skripte werden von oben nach unten ausgeführt. Agentenbasierte Anwendungen bewegen sich durch ein Graph: Jeder Knoten erledigt eine Aufgabe und gibt anschließend die Kontrolle über einen Kante weiter. Ein vereinfachtes Bild sieht so aus:

Input
   ↓
Node A
   ↓
Node B
   ↓
Node C
   ↓
Output

Jeder Knoten erhält einen gemeinsamen Zustand, aktualisiert ihn und leitet ihn weiter. Dieses Muster lässt sich auf Verzweigungen, Speicherverwaltung, Planung, Reflexion sowie mehragentige Konfigurationen erweitern – doch die erste Fähigkeit besteht darin, eine einfache Pipeline aufzubauen.

Die nachstehende Anleitung bleibt reproduzierbar: Google Colab, Open-Source-Modelle über Groq sowie eine kurze Abhängigkeitsliste. Eine lokale GPU-Infrastruktur ist nicht erforderlich. Der pädagogische Ansatz besteht darin, dass agentenbasierte Konzepte schneller umgesetzt werden können, wenn jede Zelle auf kostenloser Infrastruktur läuft – dadurch konzentriert man sich auf Zustände, Knoten und Kanten statt auf die Einrichtung von Clustern.

Abhängigkeiten installieren

!pip install langchain langgraph langchain_groq

• LangGraph erstellt das Workflow-Graph

LangChain stellt Abstraktionen für LLMs bereit

LangChain Groq kommuniziert mit Groqs Inferenz-API

Durch Ausführung der Installationen innerhalb von Colab bleibt die Umgebung wiederverwendbar und teilbar.

Erforderliche Bibliotheken importieren

from langgraph.graph import StateGraph, START, END
from typing import TypedDict
from IPython.display import Image

• StateGraph konstruiert den Workflow
• START / END kennzeichnen den Eingang und Ausgang
• TypedDict definiert die Typen des gemeinsamen Zustands
• Image rendernt ein Diagramm des kompilierten Graphen

Beispiel 1 – Einfacher sequenzieller Workflow

Starten ohne LLM, damit der Datenpfad klar erkennbar ist.

Der Zustand definieren

class BMIState(TypedDict):
    weight_kg: float
    height_m: float
    bmi: float
    category: str

Der Zustand ist ein gemeinsamer Wörterbuch, der während der Ausführung mitgeführt wird. Knoten lesen darin, schreiben Felder zu und geben die aktualisierte Karte zurück. In diesem Beispiel werden Gewicht, Größe, BMI und eine Kategorielabel verwendet.

Den Verarbeitungsknoten erstellen

# define calculate_bmi node
def calculate_bmi(state: BMIState) -> BMIState:

    weight = state['weight_kg']
    height = state['height_m']

    bmi = weight/(height**2)

    state['bmi'] = round(bmi, 2)

    return state

# define label_bmi node
def label_bmi(state: BMIState) -> BMIState:
    bmi = state['bmi']

    if bmi < 18.5:
        state["category"] = "Underweight"
    elif 18.5 <= bmi < 25:
        state["category"] = "Normal"
    elif 25 <= bmi < 30:
        state["category"] = "Overweight"
    else:
        state["category"] = "Obese"

    return state

Die Funktion liest Körpergröße und Gewicht ein, berechnet den BMI, weist eine Kategorie zu und gibt den neuen Zustand zurück. Die Funktionsweise von LangGraph ist einfach: Der aktuelle Zustand wird entgegengenommen und der nächste Zustand zurückgegeben.

Graph erstellen

# define graph
graph = StateGraph(BMIState)

# add nodes to your graph
graph.add_node('calculate_bmi', calculate_bmi)
graph.add_node('label_bmi', label_bmi)

# add edges to your graph
graph.add_edge(START, 'calculate_bmi')
graph.add_edge('calculate_bmi', 'label_bmi')
graph.add_edge('label_bmi', END)


# compile the graph
workflow = graph.compile()

# show the graph
Image(workflow.get_graph().draw_mermaid_png())

Registrieren Sie den Knoten, verbinden Sie START → Knoten → END und kompilieren Sie das Ganze. LangGraph kann den resultierenden Graphen darstellen, sodass Sie die Topologie vor der Ausführung überprüfen können.

Workflow ausführen

# execute the graph
intial_state = {'weight_kg':80, 'height_m':1.73}
final_state = workflow.invoke(intial_state)
print(final_state)

Geben Sie ein initialer Dictionary mit Gewicht und Körpergröße ein. invoke führt den Pfad aus und gibt einen Zustand zurück, der nun BMI sowie Kategorie enthält.

Beispiel 2 – Sequenzieller Workflow mit einem LLM

Tauschen Sie den Mathematik-Knoten gegen einen Aufruf eines Modells aus. Beibehalten Sie die gleiche Graphenstruktur.

Sammeln Sie sicher eine Groq-API-Schlüssel ein (durch Anfrage, nicht hardcodiert) und laden Sie anschließend ein zugängliches offenes Modell:

model = ChatGroq(model="openai/gpt-oss-120b")

LLM-Zustand definieren

# create a state
class LLMState(TypedDict):
    question: str
    answer: str

Die Felder reduzieren sich auf eine Frage und eine Antwort – die Zustandsstruktur folgt dem Workflow, nicht einem globalen Schema.

Erstellen Sie den LLM-Node

def llm_qa(state: LLMState) -> LLMState:

    # extract the question from state
    question = state['question']

    # form a prompt
    prompt = f'Answer the following question {question}'

    # ask that question to the LLM
    answer = model.invoke(prompt).content

    # update the answer in the state
    state['answer'] = answer

    return state

Lesen Sie die Frage, rufen Sie das Chat-Modell auf, speichern Sie die Antwort und geben Sie den Zustand zurück. Strukturell identisch mit dem BMI-Node; nur die internen Abläufe ändern sich.

Bauen Sie den LLM-Workflow auf

# create our graph
graph = StateGraph(LLMState)

# add nodes
graph.add_node('llm_qa', llm_qa)

# add edges
graph.add_edge(START, 'llm_qa')
graph.add_edge('llm_qa', END)

# compile
workflow = graph.compile()

Fügen Sie den Node hinzu, verbinden Sie die Kanten und kompilieren Sie. Das Gerüst bleibt absichtlich vertraut.

Stellen Sie die Frage an den LLM

# execute
intial_state = {'question': 'How far is sun from the earth?'}
final_state = workflow.invoke(intial_state)
print(final_state['answer'])

Rufen Sie das Modell mit einem Fragestring auf und geben Sie das Antwortfeld aus. Das Modell befindet sich im Graphen wie jeder andere Schritt.

Beispiel 3 – Mehrschrittiger KI-Workflow

Echte Systeme verknüpfen mehrere Überlegungsstufen miteinander. Erweitern Sie dieses Muster.

Definieren Sie den Blog-Zustand

class BlogState(TypedDict):
    title: str
    outline: str
    content: str
    rating: int

Verfolgen Sie im Laufe der Bearbeitung Titel, Überblick, Inhalt sowie eine Qualitätseinschätzung.

Erstellen Sie Workflow-Node

def create_outline(state: BlogState) -> BlogState:
    # fetch title
    title = state['title']

    # call llm gen outline
    prompt = f'Generate a detailed outline for a blog on the topic - {title}'
    outline = model.invoke(prompt).content

    # update state
    state['outline'] = outline

    return state

def create_blog(state: BlogState) -> BlogState:
    title = state['title']
    outline = state['outline']

    prompt = f'Write a short blog (less than 20 lines) on the title - {title} using the follwing outline \n {outline}'
    content = model.invoke(prompt).content
    state['content'] = content
    return state

def create_rating(state: BlogState) -> BlogState:
    title = state["title"]
    content = state["content"]

    prompt = f"""
    Rate the following blog out of 10.
    Return ONLY a single integer between 1 and 10.

    Title: {title}

    Blog:
    {content}
    """

    rating = model.invoke(prompt).content.strip()

    state["rating"] = int(rating)

    return state

Getrennte Funktionen sind für die Erstellung des Überblicks sowie für die späteren Schritte zuständig. Jeder Knoten hat eine einzige Aufgabe, damit die Diagramme lesbar bleiben.

Das vollständige Pipeline zusammenstellen

graph = StateGraph(BlogState)

# nodes
graph.add_node('create_outline', create_outline)
graph.add_node('create_blog', create_blog)
graph.add_node('create_rating', create_rating)

# edges
graph.add_edge(START, 'create_outline')
graph.add_edge('create_outline', 'create_blog')
graph.add_edge('create_blog', 'create_rating')
graph.add_edge('create_rating', END)

workflow = graph.compile()
workflow

Vereinbaren Sie die KI-Aufgaben in der richtigen Reihenfolge. Konzeptionell gesehen:

Title
   ↓
Create Outline
   ↓
Generate Blog
   ↓
Evaluate Blog
   ↓
END

Die Aufteilung der Arbeit auf verschiedene Schritte ist besser als das Hineinpacken aller Anweisungen in einen einzigen Mega-Prompt – jeder Schritt beeinflusst den nächsten.

Den Workflow ausführen

intial_state = {'title': 'Rise of AI in India'}
final_state = workflow.invoke(intial_state)
print(final_state['rating'])

Startend mit einem Titel erstellt der Prozess einen Überblick, entwirft Inhalte, bewertet die Qualität und gibt das Bewertungsfeld zurück. Das ist der Rahmen von agilen Pipelines: Mehrere Modellfähigkeiten werden hinter einem gemeinsamen Zustand koordiniert.

Betrachten Sie jedes Beispiel als Kalibrierungsschritt und nicht als wertlosen Demo-Code. Nachdem der BMI-Graph ausgeführt wurde, ändern Sie die Kategorieschwellenwerte und überprüfen Sie die zurückgegebenen Zustandsaktualisierungen. Nachdem der Graph für den einzeiligen LLM-Aufruf abgeschlossen ist, wechseln Sie die Modell-ID oder das Prompt-Muster und beobachten Sie nur die Veränderung im Verhalten dieses Knotens. Nachdem der Blog-Pipeline-Prozess abgeschlossen ist, fügen Sie einen zusätzlichen Knoten zwischen Ausgangsstruktur und Entwurf ein – vielleicht eine „Tonüberprüfung“ –, um zu erkennen, wie sich Kanten und Zustandsfelder gemeinsam entwickeln.

In dieser Phase ist Disziplin bei der Verwaltung des gemeinsamen Zustands wichtiger als clevere Prompts. Verwenden Sie lieber kleine, benannte Felder (outline, content, rating) anstelle eines riesigen messages-Blobs, solange Sie noch kein Chatverlauf benötigen. Es ist in Ordnung, eine teilweise Aktualisierung zurückzugeben, bei der unverwandte Schlüssel weggelassen werden, sofern die Graph-Verknüpfungssemantik dies zulässt; das wichtige Verhalten ist, dass die Knoten ausreichend rein bleiben, um sie mit einem manuell erstellten Dictionary testen zu können.

In Colab kann die Fehlerbehandlung minimal bleiben, muss aber dennoch absichtlich gestaltet werden: Wenn der Groq-Aufruf fehlschlägt, sollte die Ausnahme sichtbar gemacht werden anstatt im Knoten verschluckt zu werden. Frühe Tutorials, die Fehler verbergen, vermitteln den falschen Ansatz für spätere Graphen mit Tool-Aufrufen, bei denen ein fehlgeschlagener Tool zu einem expliziten Zweig werden muss.

Zusätzlich sollte der kompilierte Graph jedes Mal neu gezeichnet werden, wenn ein Kante hinzugefügt wird. Eine visuelle Überprüfung erkennt versehentliche Zyklen sowie vergessene END-Verbindungen schneller als das Lesen von add_edge-Aufrufen. Sobald das Bild der Geschichte entspricht, die man erzählen wollte, wird invoke zu einem Überprüfungsschritt statt zu einer Debugging-Sitzung.

Was diese Beispiele zeigen

Drei schrittweise aufsteigende Arbeitsabläufe umfassen das grundlegende Vokabular:

1. Einfache sequenzielle Python-Logik auf einem gemeinsamen Zustand
2. Ein einziger LLM-Knoten innerhalb desselben Graphenmusters

Eine mehrstufige KI-Kette mit mehreren zusammenarbeitenden Knoten

Weiterentwicklung: gemeinsamer Zustand, Knoten, Kanten, Orchestrierung, Integration von LLMs sowie schrittweises Reasoning. Die gleichen Konzepte unterstützen später auch Tools, Speicher, Verzweigungen sowie kooperierende Agenten.

Was kommt als Nächstes

In späteren Teilen einer agierenden KI mit Python-Path werden in der Regel bedingte Kanten, parallele Verzweigungen, Genehmigungsschritte durch Menschen, Aufrufe von Tools, RAG, Mehr-Agenten-Strukturen sowie Produktionsverpackungen hinzugefügt – weiterhin mit dem Ziel von für Colab geeigneten, offenen Modellkonfigurationen.

Vollständiger Quelltext

Das begleitende Colab-Notebook ist unter dem Agentic-AI-Projekt auf GitHub veröffentlicht (Datei Workflow_in_LangGraph.ipynb). Öffnen Sie dieses Notebook in Colab, führen Sie die Zellen von oben nach unten aus und erstellen Sie anschließend eigene Knoten für Ihre Aufgaben.