Skip to content

Theorie: Von LangChain zu LangGraph

Was ist LangGraph?

LangGraph erweitert LangChain um zustandsbasierte Workflows und flexible Routing-Logik. Während klassische LangChain-Ketten sequenziell ablaufen, ermöglicht LangGraph die Erstellung von Graphen mit Knoten und Kanten, die dynamisch Entscheidungen treffen können.

Warum LangGraph für Finance-Anwendungen?

Dies ist besonders wertvoll für Finance-Anwendungen, wo verschiedene Risikotypen unterschiedliche Behandlungsstrategien erfordern:

  • Flexible Routing-Logik basierend auf Risikotypen
  • Zustandsbasierte Workflows statt sequenzieller Abläufe
  • Dynamische Entscheidungen je nach Kontext
  • Komplexe Verzweigungen für verschiedene Finanzszenarien

Kernkonzepte

StateGraph-Architektur

StateGraph definiert die Struktur eines Workflows mit folgenden Komponenten:

  • Knoten (Nodes): Führen spezialisierte Aufgaben aus (z.B. classify_risk, handle_risk)
  • Kanten (Edges): Steuern den Workflow basierend auf Zuständen und Bedingungen
  • Zustand (State): Gemeinsamer Zustand zwischen allen Knoten
  • Entry/Exit Points: Start- und Endpunkte des Workflows

LangGraph vs. LangChain

Aspekt LangChain LangGraph
Struktur Sequenzielle Ketten Flexible Graphen
Routing Linear Conditional/Dynamic
Zustand Zwischen Schritten verloren Persistent über gesamten Workflow
Verzweigungen Begrenzt Unbegrenzt komplex
Use Cases Einfache Pipelines Komplexe Entscheidungslogik

Grundlegendes Code-Beispiel

from typing import TypedDict, Optional
from langgraph.graph import StateGraph, END

# 1. Zustandstyp definieren
class RiskState(TypedDict):
    question: str
    risk_level: Optional[str]
    response: Optional[str]

# 2. Knoten-Funktionen
def classify_risk(state):
    question = state["question"]
    # Risikoklassifizierung implementieren
    risk_level = "high" if "kredit" in question.lower() else "low"
    return {"risk_level": risk_level}

def handle_risk(state):
    risk_level = state["risk_level"]
    if risk_level == "high":
        response = "Detaillierte Risikoanalyse erforderlich"
    else:
        response = "Standard-Risikobewertung ausreichend"
    return {"response": response}

# 3. Graph aufbauen
graph = StateGraph(RiskState)
graph.add_node("classify", classify_risk)
graph.add_node("handle", handle_risk)
graph.add_edge("classify", "handle")
graph.add_edge("handle", END)
graph.set_entry_point("classify")

# 4. Kompilieren und ausführen
app = graph.compile()
result = app.invoke({"question": "Kreditrisiko Analyse"})

Conditional Routing

Für komplexere Szenarien können Sie bedingte Routing-Logik implementieren:

def route_to_expert(state):
    risk_type = state["risk_type"]

    if risk_type == "financial":
        return "financial_expert"
    elif risk_type == "compliance":
        return "compliance_expert"
    else:
        return "general_handler"

# Conditional Edge hinzufügen
graph.add_conditional_edges(
    "classify_risk_type",
    route_to_expert,
    {
        "financial_expert": "financial_expert",
        "compliance_expert": "compliance_expert", 
        "general_handler": "general_handler"
    }
)

Praktische Anwendungsfälle

1. Risk Management Workflows

  • Automatische Risikoklassifizierung
  • Routing zu spezialisierten Experten
  • Eskalationslogik bei kritischen Risiken

2. Compliance-Prozesse

  • Dokumentenprüfung mit mehrstufiger Genehmigung
  • Automatische Compliance-Checks
  • Audit-Trail Generierung

3. Financial Analysis Pipelines

  • Multi-Step Finanzanalysen
  • Bedingte Berechnungen basierend auf Datenqualität
  • Integrierte Validierung und Qualitätskontrolle

Best Practices

1. State Design

  • Definieren Sie klare, typisierte State-Strukturen
  • Verwenden Sie Optional-Felder für schrittweise Datenerweiterung
  • Dokumentieren Sie State-Übergänge

2. Node Implementation

  • Halten Sie Knoten-Funktionen fokussiert und testbar
  • Implementieren Sie Fehlerbehandlung in jedem Knoten
  • Loggen Sie wichtige State-Änderungen

3. Graph Structure

  • Starten Sie mit einfachen linearen Flows
  • Erweitern Sie schrittweise um Conditional Routing
  • Visualisieren Sie komplexe Graphen für bessere Verständlichkeit

Debugging & Monitoring

Graph-Struktur anzeigen

# Textuelle Darstellung des Graphen
app.get_graph().print_ascii()

# Knoten und Verbindungen auflisten
print("Nodes:", app.get_graph().nodes)
print("Edges:", app.get_graph().edges)

State-Tracking

def debug_node(state):
    print(f"Current state: {state}")
    return state

# Debug-Knoten in Graph einfügen
graph.add_node("debug", debug_node)

Streaming für Real-time Updates

# Stream von Updates für Live-Monitoring
for update in app.stream({"question": "Risk analysis"}):
    print(f"Update: {update}")

Weiterführende Ressourcen

💡 Nächster Schritt: Implementieren Sie Ihren ersten LangGraph mit den praktischen Übungen