LangGraph Einsteiger

Multi-Agent-Systeme und Workflows mit LangGraph

---

Inhaltsverzeichnis

1. Kurzüberblick: Warum LangGraph?
2. Das kleinstmögliche funktionierende Beispiel
   1. State definieren
   2. Ein einzelner Node
   3. Graph bauen
   4. Graphen visualisieren
   5. Ausführen
3. Die Grundidee: Workflows als State Machine
4. State sauber definieren (vertieft)
5. Nodes: Bausteine des Workflows
   1. Typ 1: LLM-Node
   2. Typ 2: Tool-Node
   3. Typ 3: Agent-Node
6. Edges & Conditional Routing
   1. Lineare Edges
   2. Bedingtes Routing
   3. Typische Muster
7. Streaming: Schritte sichtbar machen
8. Checkpointing & Sessions
9. Human-in-the-Loop (Approval & Formulare)
   1. Interrupt
   2. Weiterführen
10. Multi-Agent-Workflows (Fortgeschritten)
    1. Agenten definieren
    2. Supervisor
11. Abgrenzung zu verwandten Dokumenten

---

Kurzüberblick: Warum LangGraph?

LangChain bietet Modelle, Tools und einfache Agenten. LangGraph baut darauf auf und ermöglicht:

• Abläufe in mehreren Schritten
• Bedingte Entscheidungswege
• Mehrere Agenten (Rollen)
• Sitzungen, die wieder aufgenommen werden können
• Human-in-the-Loop

Ein Workflow besteht aus:

• State: zentrale Daten
• Nodes: Bearbeitungsschritte
• Edges: Ablaufsteuerung

Ein minimales Diagramm:

graph LR
    A[LangChain] --> B[LangGraph]
    B --> C[Multi-Step Workflows]
    B --> D[Conditional Routing]
    B --> E[Multi-Agent Systems]
    B --> F[Checkpointing]
    B --> G[Human-in-Loop]

    style B fill:#10a37f
    style A fill:#0066cc

Damit ist sofort klar: LangGraph strukturiert Workflows, anstatt alles in ein einzelnes LLM-Prompt zu packen.

---

Das kleinstmögliche funktionierende Beispiel

Der schnellste Weg zum Verständnis ist ein Mini-Workflow.

State definieren

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class ChatState(TypedDict):
    messages: Annotated[list, add_messages]
    step: int

Ein einzelner Node

from langchain.chat_models import init_chat_model
llm = init_chat_model("openai:gpt-4o-mini", temperature=0.0)

def agent_node(state: ChatState) -> ChatState:
    response = llm.invoke(state["messages"])
    return {"messages": [response], "step": state["step"] + 1}

Graph bauen

from langgraph.graph import StateGraph, START, END

g = StateGraph(ChatState)
g.add_node("agent", agent_node)
g.add_edge(START, "agent")
g.add_edge("agent", END)

graph = g.compile()

Graphen visualisieren

from IPython.display import Image

display(Image(graph.get_graph().draw_mermaid_png()))

Dieser minimale Workflow sieht so aus:

flowchart LR
    START([START]) --> AGENT[agent_node]
    AGENT --> END([END])

    style START fill:#90EE90
    style END fill:#FFB6C1
    style AGENT fill:#87CEEB

Ausführen

from langchain_core.messages import HumanMessage

initial_state = {"messages": [HumanMessage(content="Was ist LangGraph?")], "step": 0}
result = graph.invoke(initial_state)
result

Ergebnis: Ein vollständiger Einsteiger-Workflow, bevor irgendein abstraktes Konzept erklärt wurde.

---

Die Grundidee: Workflows als State Machine

Nachdem Einsteiger ein funktionsfähiges Beispiel gesehen haben, kann das Konzept erklärt werden:

• Ein Workflow besteht aus klar definierten Schritten (Nodes).
• Der Zustand wird in einem State gespeichert.
• Edges bestimmen die Reihenfolge.
• Reducer wie add_messages fügen Informationen intelligent zusammen.

stateDiagram-v2
    [*] --> State: Initialize
    State --> Node1: Edge
    Node1 --> State: Update (Reducer)
    State --> Node2: Edge
    Node2 --> State: Update (Reducer)
    State --> [*]: Complete

    note right of State
        messages: []
        step: 0
    end note

Kurz: Nodes sind Funktionen – Edges sind der Ablauf.

---

State sauber definieren (vertieft)

class ChatState(TypedDict):
    messages: Annotated[list, add_messages]
    step: int
    # erweiterbar: approved: bool, analysis: str

Prinzipien:

• Nur speichern, was später benötigt wird.
• Typisierung unterstützt Verständnis und Fehlersuche.
• Reducer sorgen dafür, dass Listen (z. B. Nachrichten) korrekt gemergt werden.

---

Nodes: Bausteine des Workflows

Nodes sollen klein, fokussiert und deterministisch sein.

Ein Node ist immer eine Python-Funktion — was sich unterscheidet, ist der Inhalt:

Node-Typ	Inhalt der Funktion	Typischer Einsatz
LLM-Node	direkter LLM-Aufruf	Antworten, Zusammenfassungen
Tool-Node	Tool-Ausführung	Suche, Berechnungen, APIs
Agent-Node	vollständiger Agent (create_react_agent)	komplexe Teilaufgaben mit eigenem Tool-Loop

Typ 1: LLM-Node

def summarize_node(state: ChatState) -> ChatState:
    text = "\n".join([m.content for m in state["messages"]])
    summary = llm.invoke([{"role": "user", "content": f"Fasse zusammen: {text}"}])
    return {"messages": [summary]}

Typ 2: Tool-Node

from langchain_core.tools import tool

@tool
def suche(query: str) -> str:
    """Führt eine Websuche durch."""
    return f"Ergebnis für: {query}"

def tool_node(state: ChatState) -> ChatState:
    result = suche.invoke({"query": state["messages"][-1].content})
    return {"messages": [{"role": "tool", "content": result}]}

Typ 3: Agent-Node

Ein Node kann intern einen vollständigen Agenten ausführen — inklusive eigenem Tool-Loop:

from langgraph.prebuilt import create_react_agent

research_agent = create_react_agent(
    model=llm,
    tools=[suche],
    prompt="Du bist ein Research-Spezialist. Recherchiere gründlich.",
)

def research_node(state: ChatState) -> ChatState:
    result = research_agent.invoke({"messages": state["messages"]})
    return {"messages": result["messages"]}

Hinweis: Der Agent-Node ist das Muster hinter Supervisor-Architekturen (M20, M21). Der Supervisor ruft research_node, writer_node etc. auf — jeder Node kapselt intern einen vollständigen Agenten.

---

Edges & Conditional Routing

Nun erst wird Routing eingeführt – nachdem Einsteiger Nodes und State kennen.

Lineare Edges

g.add_edge(START, "agent")
g.add_edge("agent", END)

Bedingtes Routing

def tool_node(state: ChatState):
    result_message = ...  # echtes Tool
    return {"messages": [result_message]}

def routing_after_agent(state: ChatState) -> str:
    last_msg = state["messages"][-1]
    if getattr(last_msg, "tool_calls", None):
        return "tools"
    return END

g.add_node("tools", tool_node)
g.add_conditional_edges(
    "agent",
    routing_after_agent,
    {"tools": "tools", END: END},
)
g.add_edge("tools", "agent")

Visualisierung des Tool-Loops:

flowchart TB
    START([START]) --> AGENT[Agent Node]
    AGENT --> COND{Tool Calls?}
    COND -->|Yes| TOOLS[Tool Node]
    COND -->|No| END([END])
    TOOLS --> AGENT

    style START fill:#90EE90
    style END fill:#FFB6C1
    style COND fill:#FFD700
    style TOOLS fill:#FFA500
    style AGENT fill:#87CEEB

Typische Muster

• Schlüsselwort-Trigger
• Unsicherheitsanalyse
• Routing nach Tool-Feedback
• Wiederholschleifen (Retry)

---

Streaming: Schritte sichtbar machen

Streaming ist ein wichtiges Werkzeug für das Verständnis.

for event in graph.stream(initial_state, {"configurable": {"thread_id": "demo"}}, stream_mode="updates"):
    print(event)

Streaming-Prozess:

sequenceDiagram
    autonumber
    participant User
    participant Graph
    participant Node1
    participant Node2

    User->>Graph: invoke(state)
    Graph->>Node1: process
    Node1-->>User: stream update 1
    Graph->>Node2: process
    Node2-->>User: stream update 2
    Graph-->>User: final result

Streaming-Varianten:

• updates: nur Änderungen
• values: vollständiger State
• messages: nur neue Nachrichten

Empfehlung für Einsteiger: updates.

---

Checkpointing & Sessions

Checkpointing ermöglicht:

• längerfristige Workflows
• Resume nach Unterbrechung
• stabile Interaktion

from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
graph = g.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "session-01"}}
result1 = graph.invoke(initial_state, config)

Später:

result2 = graph.invoke(None, config)  # setzt fort

Session-Management mit Checkpointing:

stateDiagram-v2
    [*] --> Session1: invoke(state, thread_id)
    Session1 --> Checkpoint: save state
    Checkpoint --> Session1: resume
    Session1 --> Session2: continue later
    Session2 --> [*]: complete

    note right of Checkpoint
        MemorySaver (Dev)
        SQLite (Staging)
        Postgres (Production)
    end note

Hinweise:

• Optimale Einstiegsvariante: MemorySaver.
• Für produktive Systeme: SQLite/Postgres.
• Graphänderungen können Sessions invalidieren.

---

Human-in-the-Loop (Approval & Formulare)

Human-in-the-Loop ist ein wichtiges Konzept – aber erst an dieser Stelle sinnvoll.

Interrupt

from langgraph.types import interrupt

def approval_node(state: ChatState) -> ChatState:
    decision = interrupt("Aktion ausführen? yes/no")
    return {"approved": decision == "yes"}

Weiterführen

from langgraph.types import Command
result = graph.invoke(Command(resume="yes"), config)

Human-in-the-Loop Pattern:

flowchart TB
    START([START]) --> NODE1[Process Data]
    NODE1 --> APPROVAL[👤 Approval Node]
    APPROVAL --> INTERRUPT{interrupt}
    INTERRUPT -.Wait for Human.-> DECISION{Decision?}
    DECISION -->|yes| EXECUTE[Execute Action]
    DECISION -->|no| REJECT[Reject Action]
    EXECUTE --> END([END])
    REJECT --> END

    style START fill:#90EE90
    style END fill:#FFB6C1
    style APPROVAL fill:#FFA500
    style INTERRUPT fill:#FFD700
    style EXECUTE fill:#87CEEB
    style REJECT fill:#ff6b6b

Einsatzmöglichkeiten:

• sicherheitsrelevante Aktionen
• wichtige Entscheidungen
• mehrschrittige Formulareingaben

---

Multi-Agent-Workflows (Fortgeschritten)

Dieses Thema wurde bewusst ans Ende verschoben.

Agenten definieren

from langchain.agents import create_agent

research_agent = create_agent(model=llm, tools=[...], system_prompt="Research")
writer_agent   = create_agent(model=llm, tools=[...], system_prompt="Writer")

Supervisor

from langgraph.types import Command

def supervisor(state: ChatState) -> Command:
    task = state.get("task_type", "research")
    return Command(goto=f"{task}_agent")

Multi-Agent Supervisor Pattern:

graph TB
    START([User Request]) --> SUPERVISOR[Supervisor Agent]

    SUPERVISOR --> COND{Task Type?}
    COND -->|research| RESEARCH[Research Agent]
    COND -->|writing| WRITER[Writer Agent]
    COND -->|analysis| ANALYST[Analyst Agent]

    RESEARCH --> REVIEW[Quality Check]
    WRITER --> REVIEW
    ANALYST --> REVIEW

    REVIEW --> ITERCHECK{Good enough?}
    ITERCHECK -->|No| SUPERVISOR
    ITERCHECK -->|Yes| END([END])

    style SUPERVISOR fill:#10a37f
    style RESEARCH fill:#87CEEB
    style WRITER fill:#FFB6C1
    style ANALYST fill:#FFA500
    style REVIEW fill:#FFD700

Mögliche Erweiterungen:

• iterative Qualitätsprüfungen
• mehrere Worker mit Prioritäten
• automatische oder manuelle Rollenwechsel

Abgrenzung zu verwandten Dokumenten

Dokument	Inhalt
Einsteiger LangChain	Voraussetzung: Modell-Init, Tools und Agenten mit LangChain
Einsteiger ChromaDB	Vektordatenbank als RAG-Tool in LangGraph-Workflows
State Management	Konzeptionelle Tiefe hinter TypedDict und Reducer-Funktionen
Checkpointing & Persistenz	Technische Details zu MemorySaver und Thread-IDs
Human-in-the-Loop	Konzept hinter Interrupt & Resume aus Abschnitt 9

---

Version: 2.0
Stand: Januar 2026
Kurs: KI-Agenten. Verstehen. Anwenden. Gestalten.