KI-Agenten Blog
COMPUTERZUGRIFFSSYSTEM

Zustandsmaschinen und Agenten: Zuverlässige Workflows mit LangGraph

agentsarchitecturetutoriallanggraphautonomous-systems

Die meisten Agenten-Tutorials zeigen eine einfache Schleife: Claude fragen, Antwort parsen, Tool aufrufen, wiederholen. Das funktioniert für Demos. In der Produktion braucht man jedoch Determinismus, Fehlerwiederherstellung, menschliche Genehmigungsschritte und Auditierbarkeit.

LangGraph bringt Zustandsmaschinen in Agenten-Workflows. Statt einer Ad-hoc-Schleife aus if-Anweisungen erhält man einen expliziten Graphen: benannte Knoten (Logikeinheiten), typisierte Kanten (Übergänge) und ein gemeinsames Zustandsschema, das durch die gesamte Ausführung fließt.

Warum Zustandsmaschinen für Agenten?

Das Ad-hoc-Schleifen-Problem

Eine typische Agentenschleife sieht so aus:

messages = []
while True:
    response = claude.messages.create(model=..., messages=messages)
    if response.stop_reason == "end_turn":
        break
    for tool_call in get_tool_calls(response):
        result = execute_tool(tool_call)
        messages.append(tool_result(tool_call.id, result))

Das ist für zwei oder drei Tools lesbar. Fügt man fünf Tools, bedingte Pfade, einen menschlichen Genehmigungsschritt und Retry-Logik hinzu, hat man hunderte Zeilen verschachtelter Ablaufsteuerung.

Das tiefere Problem ist der implizite Zustand. In welcher Phase ist der Agent? Welche Daten hat er gesammelt? Alles lebt in messages — ein nicht typisierter Blob, den jeder Knoten liest und ergänzt, ohne erzwungenes Schema.

Zustandsmaschinen als Lösung

Eine Zustandsmaschine macht das Implizite explizit. Man definiert:

  • Knoten — diskrete Logikeinheiten, die den aktuellen Zustand empfangen, eine Aufgabe erledigen und Zustandsaktualisierungen zurückgeben.
  • Kanten — Übergänge zwischen Knoten, entweder unbedingt (A → B immer) oder bedingt.
  • Zustand — ein typisiertes Dictionary, das durch den gesamten Graphen fließt. Das Schema wird bei jedem Schritt validiert.

Wann man LangGraph NICHT verwenden sollte

Für einfache Einzel-Aufgaben ist ein direkter API-Aufruf schneller und klarer. LangGraph verwenden wenn:

  • Mehrere verschiedene Phasen in Sequenz ausgeführt werden müssen
  • Bedingtes Verzweigen basierend auf Zwischenergebnissen
  • Menschliche Interaktionsschritte
  • Fehlerwiederherstellungs- oder Retry-Logik
  • Auditierbarkeitsanforderungen

LangGraph-Grundlagen

Zustand definieren

from typing import TypedDict


class ResearchState(TypedDict):
    query: str
    research_notes: str
    draft: str
    review_feedback: str
    is_approved: bool

Knoten

import anthropic


client = anthropic.Anthropic()


def research_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"Recherchiere dieses Thema gründlich: {state['query']}"
        }]
    )
    return {"research_notes": response.content[0].text}


def draft_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"Schreibe einen Entwurf basierend auf diesen Notizen:\n{state['research_notes']}"
        }]
    )
    return {"draft": response.content[0].text}

Kanten und bedingtes Routing

from typing import Literal


def route_after_check(state: ResearchState) -> Literal["human_review", "draft"]:
    if state.get("review_feedback"):
        return "human_review"
    return "draft"


workflow.add_conditional_edges(
    "research",
    route_after_check,
    {"human_review": "review_node", "draft": "draft_node"}
)

Graphen erstellen und ausführen

from langgraph.graph import StateGraph, END


workflow = StateGraph(ResearchState)
workflow.add_node("research", research_node)
workflow.add_node("draft", draft_node)
workflow.set_entry_point("research")
workflow.add_edge("research", "draft")
workflow.add_edge("draft", END)


graph = workflow.compile()
result = graph.invoke({"query": "Was ist das MCP-Protokoll?"})
print(result["draft"])

Einen Dokumentenprüfungs-Workflow Erstellen

Zustand entwerfen

class DocumentState(TypedDict):
    document: str
    key_terms: list[str]
    compliance_issues: list[str]
    summary: str
    human_feedback: str
    is_approved: bool

Knoten implementieren

import json


def extract_terms_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": (
                "Extrahiere genau 5 Schlüsselbegriffe aus diesem Dokument. "
                "Gib sie als JSON-Array von Strings zurück.\n\n"
                f"Dokument: {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        terms = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        terms = [t.strip() for t in response.content[0].text.split("\n") if t.strip()][:5]
    return {"key_terms": terms}




def check_compliance_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": (
                "Prüfe dieses Dokument auf Compliance-Probleme. "
                "Suche nach: personenbezogenen Daten (Namen, Steuer-IDs, E-Mails, Telefonnummern), "
                "vertraulichen Kennzeichnungen, unverifizierten Behauptungen.\n\n"
                "Gib ein JSON-Array von Problembeschreibungen zurück. "
                "Gib ein leeres Array [] zurück, wenn keine Probleme gefunden werden.\n\n"
                f"Dokument: {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        issues = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        issues = []
    return {"compliance_issues": issues}




def human_review_node(state: DocumentState) -> dict:
    print("\n=== MENSCHLICHE ÜBERPRÜFUNG ERFORDERLICH ===")
    for issue in state["compliance_issues"]:
        print(f"  - {issue}")
    return {
        "human_feedback": "Überprüft und genehmigt nach Schwärzung personenbezogener Daten",
        "is_approved": True
    }




def summarize_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=256,
        messages=[{
            "role": "user",
            "content": f"Fasse dieses Dokument in genau zwei Sätzen zusammen:\n\n{state['document']}"
        }]
    )
    return {"summary": response.content[0].text, "is_approved": True}

Den Graphen zusammenstellen

def route_after_compliance(state: DocumentState) -> Literal["human_review", "summarize"]:
    if state.get("compliance_issues"):
        return "human_review"
    return "summarize"


workflow = StateGraph(DocumentState)
workflow.add_node("extract", extract_terms_node)
workflow.add_node("check", check_compliance_node)
workflow.add_node("review", human_review_node)
workflow.add_node("summarize", summarize_node)


workflow.set_entry_point("extract")
workflow.add_edge("extract", "check")
workflow.add_conditional_edges(
    "check", route_after_compliance,
    {"human_review": "review", "summarize": "summarize"}
)
workflow.add_edge("review", "summarize")
workflow.add_edge("summarize", END)


graph = workflow.compile()


result = graph.invoke({
    "document": "Patient Max Mustermann (Steuernummer: 12/345/67890) leidet an Bluthochdruck.",
    "key_terms": [], "compliance_issues": [],
    "summary": "", "human_feedback": "", "is_approved": False,
})
print(result["summary"])

Menschliche Checkpoints und Unterbrechungen

from langgraph.checkpoint.memory import MemorySaver


checkpointer = MemorySaver()
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["review"]
)


thread_config = {"configurable": {"thread_id": "doc-review-001"}}
result = graph.invoke(initial_state, config=thread_config)


current_state = graph.get_state(thread_config)
print("Probleme:", current_state.values["compliance_issues"])


graph.update_state(
    thread_config,
    {"human_feedback": "Genehmigt nach manueller Schwärzung", "is_approved": True}
)


final_result = graph.invoke(None, config=thread_config)

Persistenz und Produktionseinsatz

from langgraph.checkpoint.postgres import PostgresSaver


with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    graph = workflow.compile(checkpointer=checkpointer)


import asyncio


async def batch_process(documents: list[str]) -> list[dict]:
    return await asyncio.gather(*[
        graph.ainvoke({"document": doc, "key_terms": [], "compliance_issues": [],
                       "summary": "", "human_feedback": "", "is_approved": False})
        for doc in documents
    ])

Häufige Muster und Fehler

  • Fan-Out / Fan-In: Mit Send mehrere Knoten parallel ausführen und Ergebnisse zusammenführen.
  • Zu komplexes bedingtes Routing: Bei zu vielen Verzweigungen in Teilgraphen aufteilen.
  • Zustandsüberladung: Referenzen (Datenbank-IDs, S3-Schlüssel) statt großer Artefakte speichern.
  • Knoten isoliert testen: Da Knoten einfache Funktionen sind, können sie direkt unit-getestet werden.
def test_extract_terms_node():
    state = {
        "document": "Der schnelle braune Fuchs springt über den faulen Hund.",
        "key_terms": [], "compliance_issues": [],
        "summary": "", "human_feedback": "", "is_approved": False,
    }
    result = extract_terms_node(state)
    assert "key_terms" in result
    assert isinstance(result["key_terms"], list)

LangGraph ersetzt Ad-hoc-Schleifen durch explizite, debuggbare und fortsetzbare Workflows. Mit einem linearen Graphen beginnen (A → B → C → ENDE). Eine bedingte Kante hinzufügen, wenn Verzweigungen nötig sind. Einen menschlichen Checkpoint hinzufügen, wenn Genehmigungen benötigt werden. Die meisten Agenten-Produktions-Workflows benötigen genau diese drei Muster.

Verwandte Artikel