Skip to content

Übungen - LangGraph & Advanced Workflows

🎯 Lernziele

In diesen Übungen werden Sie: - Ihren ersten LangGraph für Risikoklassifizierung erstellen - Ein komplexes Routing-System mit spezialisierten Expertenknoten implementieren - Eine Compliance-Workflow mit mehrstufiger Genehmigung aufbauen - Workflows mit Mermaid-Diagrammen visualisieren

Aufgabe 1: Grundlagen - Einfacher Risk-Classifier

Ziel: Erstellen Sie Ihren ersten LangGraph für die Risikoklassifizierung

Starter-Code (Grundgerüst)

Beginnen Sie mit diesem Template und implementieren Sie die fehlenden Funktionen:

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

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

def classify_risk(state):
    # TODO: Implementieren Sie die Klassifizierungslogik
    # Hinweis: Analysieren Sie die Frage und bestimmen Sie das Risikolevel
    pass

def handle_risk(state):
    # TODO: Implementieren Sie die Behandlungslogik
    # Hinweis: Basierend auf risk_level eine angemessene Antwort generieren
    pass

# 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")

app = graph.compile()

Implementierungshinweise

  1. classify_risk() Funktion:
  2. Analysieren Sie die Eingabefrage nach Schlüsselwörtern
  3. Klassifizieren Sie als "high", "medium" oder "low" risk

  4. Rückgabe: {"risk_level": "high"}

  5. handle_risk() Funktion:

  6. Erstellen Sie unterschiedliche Antworten basierend auf dem Risikolevel
  7. Rückgabe: {"response": "Detaillierte Analyse erforderlich"}

Testfälle

# Test Ihres Risk-Classifiers
test_cases = [
    {"question": "Kreditrisiko bei Immobilienfinanzierung"},
    {"question": "Routine Kontobewegung prüfen"},
    {"question": "Verdächtiger Geldtransfer erkannt"}
]

for test in test_cases:
    result = app.invoke(test)
    print(f"Input: {test['question']}")
    print(f"Risk Level: {result['risk_level']}")
    print(f"Response: {result['response']}\n")

Vollständiger Starter-Code: simple_langgraph_example.py

Aufgabe 2: Routing-Workflow für Finanzrisiken

Ziel: Implementierung eines komplexen Routing-Systems mit spezialisierten Expertenknoten

Arbeitsanweisung

Nutzen Sie das Jupyter Notebook für die schrittweise Entwicklung: LangGraph Übungen Notebook

Zu implementierende Funktionen

  1. classify_risk_type() - Risikotyp bestimmen (Marktrisiko, Kreditrisiko, etc.)
  2. financial_expert() - Finanzrisiko-Analyse
  3. compliance_expert() - Compliance-Prüfung
  4. route_to_expert() - Dynamisches Routing

Erweiterte StateGraph-Struktur

class AdvancedRiskState(TypedDict):
    question: str
    risk_type: Optional[str]
    risk_level: Optional[str]
    expert_analysis: Optional[str]
    final_recommendation: Optional[str]

def classify_risk_type(state):
    # TODO: Bestimmen Sie den Risikotyp
    # "market", "credit", "operational", "compliance"
    pass

def financial_expert(state):
    # TODO: Spezialisierte Finanzrisikoanalyse
    pass

def compliance_expert(state):
    # TODO: Compliance-spezifische Prüfung
    pass

def route_to_expert(state):
    # TODO: Routing-Logik implementieren
    risk_type = state["risk_type"]
    if risk_type == "market":
        return "financial_expert"
    elif risk_type == "compliance":
        return "compliance_expert"
    # ... weitere Routing-Regeln

Template: minimal_financial_graph.py

Aufgabe 3: Compliance-Workflow mit mehrstufiger Genehmigung

Ziel: Automatisierte Dokumentenprüfung mit Eskalationslogik

Workflow-Beschreibung

  1. Dokumentenanalyse - Automatische Prüfung eingehender Dokumente
  2. Risk-Assessment - Bewertung des Compliance-Risikos
  3. Approval-Routing - Weiterleitung basierend auf Risikostufe:
  4. Low Risk → Automatische Genehmigung
  5. Medium Risk → Manager Approval
  6. High Risk → Senior Management + Legal Review

Advanced Features

  • Multi-Level-Approval: Verschiedene Genehmigungsebenen
  • Eskalationslogik: Automatische Weiterleitung bei Zeitüberschreitung
  • Audit-Trail: Vollständige Dokumentation aller Schritte

Erweiterte Implementierung: financial_compliance_graph.py

Aufgabe 4: Workflow-Visualisierung

Ziel: Erstellen Sie Mermaid-Diagramme Ihrer Workflows für bessere Verständlichkeit

Visualisierungs-Tools

Verfügbare Tools: - mermaid_risk_graph.py - Mermaid-Diagramm Generierung - langgraph_visualization.py - Graph-Darstellung mit verschiedenen Formaten

Beispiel-Visualisierung

def create_risk_workflow_diagram():
    mermaid_code = """
    graph TD
        A[Start: Risk Question] --> B[Classify Risk Type]
        B --> C{Risk Type?}
        C -->|Market| D[Financial Expert]
        C -->|Credit| E[Credit Expert]
        C -->|Compliance| F[Compliance Expert]
        D --> G[Generate Report]
        E --> G
        F --> G
        G --> H[End: Final Recommendation]
    """
    return mermaid_code

# Diagramm rendern und speichern
diagram = create_risk_workflow_diagram()
save_mermaid_diagram(diagram, "risk_workflow.png")

🛠️ Setup & Voraussetzungen

Installation

# Alle Abhängigkeiten installieren
pip install -r requirements.txt

# Zusätzliche Pakete für Visualisierung
pip install jupyter matplotlib graphviz pillow

API-Konfiguration

  1. OpenAI API-Schlüssel erforderlich
  2. Konfiguration in credentials.yml oder Environment Variables
  3. Kostenlose Credits verfügbar unter platform.openai.com

Arbeitsumgebung

  • Empfohlen: Jupyter Notebook für iterative Entwicklung
  • Alternative: VS Code mit Python Extension
  • Testing: Verwenden Sie bereitgestellte Testdaten

📋 Erfolgskriterien

Aufgabe 1 - Risk Classifier

  • [ ] Funktionale classify_risk() und handle_risk() Funktionen
  • [ ] Korrekte Risikoklassifizierung für Testfälle
  • [ ] StateGraph kompiliert und läuft ohne Fehler

Aufgabe 2 - Routing Workflow

  • [ ] Conditional Routing basierend auf Risikotyp implementiert
  • [ ] Spezialisierte Expertenfunktionen funktional
  • [ ] Komplexer StateGraph mit mehreren Pfaden

Aufgabe 3 - Compliance Workflow

  • [ ] Multi-Level Approval Logic implementiert
  • [ ] Eskalationslogik für verschiedene Risikostufen
  • [ ] Audit-Trail Funktionalität

Aufgabe 4 - Visualisierung

  • [ ] Mermaid-Diagramm erfolgreich generiert
  • [ ] Graph-Struktur visuell verständlich dargestellt
  • [ ] Export in verschiedene Formate (PNG, SVG)

💡 Tipps & Hilfestellungen

Einstiegshilfen

  • Beginnen Sie mit dem einfachsten Beispiel und erweitern Sie schrittweise
  • Nutzen Sie print()-Statements um den State-Flow zu verfolgen
  • Testen Sie jeden Knoten einzeln, bevor Sie den kompletten Graph ausführen

Debugging

# Graph-Struktur anzeigen
app.get_graph().print_ascii()

# State zwischen Knoten verfolgen
def debug_node(state):
    print(f"Current state: {state}")
    return state

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

Häufige Probleme

  • API-Fehler: OpenAI API-Schlüssel und Guthaben prüfen
  • Import-Fehler: Alle Pakete aus requirements.txt installieren
  • Graph-Visualisierung: Internetverbindung für Mermaid-API erforderlich

📚 Weiterführende Ressourcen

💡 Nächster Schritt: Nach Abschluss der Übungen schauen Sie sich die Musterlösungen an