Lektion 3 von 6·11 Min Lesezeit

LangGraph Grundlagen

LangGraph ist LangChains Framework für komplexe, zustandsbehaftete Agent-Workflows. Während LCEL für lineare Chains reicht, ermöglicht LangGraph Zyklen, bedingte Verzweigungen und persistenten State — essenziell für Production-Agents.

Warum LangGraph?

Einfache Chains sind linear: Input → Verarbeitung → Output. Aber reale Agent-Workflows brauchen:

  • Zyklen: Der Agent versucht etwas, prüft das Ergebnis und versucht es erneut
  • Bedingte Logik: Je nach Ergebnis wird ein anderer Pfad gewählt
  • Persistenter State: Zustand wird über mehrere Schritte bewahrt
  • Human-in-the-Loop: Menschen können an definierten Punkten eingreifen

State Graph Konzept

Ein LangGraph-Graph besteht aus drei Elementen:

1. State

Der gemeinsame Zustand, der durch den Graphen fließt:

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

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    current_step: str
    results: dict

2. Nodes

Funktionen, die den State verarbeiten und modifizieren:

def research_node(state: AgentState) -> dict:
    # Research durchführen
    results = search_tool.invoke(state["messages"][-1].content)
    return {"results": {"research": results}, "current_step": "analyze"}

3. Edges

Verbindungen zwischen Nodes — statisch oder bedingt:

from langgraph.graph import StateGraph, END

graph = StateGraph(AgentState)
graph.add_node("research", research_node)
graph.add_node("analyze", analyze_node)
graph.add_node("respond", respond_node)

graph.add_edge("research", "analyze")
graph.add_conditional_edges("analyze", route_function, {
    "needs_more": "research",   # Zyklus zurück
    "ready": "respond"          # Weiter zum Antworten
})
graph.add_edge("respond", END)

Conditional Routing

Bedingte Kanten erlauben dynamische Entscheidungen:

def route_function(state: AgentState) -> str:
    if state["results"].get("confidence", 0) < 0.8:
        return "needs_more"
    return "ready"

Checkpointing

LangGraph speichert den State nach jedem Node. Bei Fehlern können Sie zum letzten Checkpoint zurückrollen:

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

Human-in-the-Loop

Definieren Sie Interrupt-Punkte, an denen menschliche Genehmigung erforderlich ist:

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["send_email", "delete_record"]
)

Praxis-Tipp: Zeichnen Sie Ihren Workflow als Flussdiagramm, bevor Sie Code schreiben. Jeder Kasten wird ein Node, jeder Pfeil eine Edge. LangGraph macht Ihre Architektur explizit — nutzen Sie das.