OpenFang: Leitfaden zur Erstellung automatisierter Workflows – Vom Einzel-Agenten zum Agenten-Team

Slug: openfang-workflow-automation-guide Kategorie: usage-guides (Anleitungen) Ziel-Keywords: OpenFang Workflow, OpenFang Automatisierungsprozesse, OpenFang Multi-Agenten-Orchestrierung Suchintention: Informationell (Benutzer möchte komplexe Agenten-Automatisierungsprozesse aufbauen) Ziel-Wortzahl: ~2000 Wörter Sprache: de

Warum benötigen Sie Workflows?

Die Aufgaben, die ein einzelner Agent erledigen kann, sind begrenzt. Wenn Sie End-to-End-Prozesse wie "Thema recherchieren → Artikel schreiben → Fakten prüfen → veröffentlichen → Feedback überwachen" benötigen, müssen Sie mehrere Agenten zu einem Workflow orchestrieren.

Die Workflow-Engine von OpenFang unterstützt drei Orchestrierungsmodi: serielle Pipelines (Pipeline), parallele Verteilung (Fan-out) und bedingte Verzweigungen (Conditional). Sie können Hands und Agenten wie Legosteine zu komplexen Automatisierungsprozessen kombinieren.

Dieser Artikel führt Sie anhand von vier praktischen Beispielen von einfachen zu komplexen Workflows in OpenFang.

Kernkonzepte der Workflows

Bevor wir in die Beispiele eintauchen, hier einige Kernkonzepte:

Konzept	Beschreibung
Task	Die kleinste Ausführungseinheit eines Workflows; ein Task entspricht einem Hand oder einem Agenten
Pipeline	Verbindet mehrere Tasks; der Output des vorherigen Tasks wird automatisch zum Input des nächsten
Fan-out	Verteilt eine Aufgabe an mehrere Agenten zur parallelen Verarbeitung und führt die Ergebnisse zusammen
Trigger	Ereignisquelle, die den Workflow startet: Zeitplan, Webhook, Dateiänderung, Nachrichtenbefehl
Context	Geteilter Status, der zwischen Workflow-Schritten übergeben wird, ähnlich einem Variablen-Scope in Programmiersprachen

Beispiel 1: Tägliches Branchen-Briefing (Einfache Pipeline)

Dies ist ein einfacher Pipeline-Workflow, der jeden Morgen automatisch ein Branchen-Briefing erstellt:

[[workflows]]name = "daily_industry_briefing"description = "Erstellt jeden Morgen um 8 Uhr ein Briefing zu den Branchen-News von gestern"[workflows.trigger]type = "schedule"cron = "0 8 * * 1-5"             # Werktags um 8 Uhr morgens[[workflows.tasks]]id = "monitor"hand = "collector"config = { keywords = ["AI agent", "Rust framework"], sources = ["news", "github"] }timeout = 300[[workflows.tasks]]id = "summarize"hand = "researcher"config = { depth = "shallow", focus = "summary" }depends_on = ["monitor"]timeout = 600[[workflows.tasks]]id = "notify"channel = "slack"config = { channel = "#industry-news", format = "rich_text" }depends_on = ["summarize"][workflows.output]format = "slack_message"save_to = "workspace/briefings/{{date}}.md"

Der Workflow-Ablauf: Collector sammelt → Researcher fasst zusammen → Slack-Benachrichtigung. depends_on definiert die Abhängigkeiten zwischen den Tasks.

Beispiel 2: Wettbewerbsüberwachung (Parallel + Aggregation)

Dieser Workflow überwacht gleichzeitig mehrere Wettbewerber und fasst die Ergebnisse nach der parallelen Erfassung zusammen:

[[workflows]]name = "competitor_monitor"description = "Überwacht 5 Wettbewerber-Websites parallel und sendet bei Änderungen eine Warnung"[workflows.trigger]type = "schedule"cron = "0 */6 * * *"# Stage 1: Parallele Erfassung (Fan-out)[[workflows.tasks]]id = "check_competitor_a"hand = "collector"config = { target = "competitor-a.com", mode = "diff" }[[workflows.tasks]]id = "check_competitor_b"hand = "collector"config = { target = "competitor-b.com", mode = "diff" }[[workflows.tasks]]id = "check_competitor_c"hand = "collector"config = { target = "competitor-c.com", mode = "diff" }# Stage 2: Aggregation (Warten auf Abschluss aller Erfassungen)[[workflows.tasks]]id = "aggregate"hand = "researcher"config = { action = "merge_and_prioritize" }depends_on = ["check_competitor_a", "check_competitor_b", "check_competitor_c"]# Stage 3: Bedingte Prüfung — Benachrichtigung nur bei wichtigen Änderungen[[workflows.tasks]]id = "check_significance"hand = "predictor"config = { action = "assess_impact" }depends_on = ["aggregate"]# Stage 4: Bedingte Benachrichtigung[[workflows.tasks]]id = "alert_high"channel = "telegram"config = { chat_id = "-100xxx", priority = "high" }depends_on = ["check_significance"]condition = "ctx.significance_score > 0.7"[[workflows.tasks]]id = "alert_low"channel = "email"config = { to = "[email protected]", priority = "low" }depends_on = ["check_significance"]condition = "ctx.significance_score <= 0.7"

Die Schlüsselmuster hier sind:

Fan-out: Die 5 Erfassungs-Tasks in Stage 1 laufen parallel, ohne depends_on.
Barrier: Der Aggregate-Task in Stage 2 wartet auf den Abschluss aller 5 Erfassungs-Tasks.
Conditional: Stage 4 wählt basierend auf dem Wichtigkeits-Score unterschiedliche Benachrichtigungskanäle.

Beispiel 3: Content-Produktions-Pipeline (Multi-Agenten-Kollaboration)

Dies ist das klassische Szenario für Multi-Agenten-Kollaboration: KI schreibt, KI prüft, KI erstellt Bilder:

[[workflows]]name = "content_pipeline"description = "Vollautomatische Content-Produktion vom Thema bis zur Veröffentlichung"[workflows.trigger]type = "webhook"endpoint = "/webhook/content/new"secret = "${WEBHOOK_SECRET}"# Stage 1: Recherche[[workflows.tasks]]id = "research"hand = "researcher"config = {  depth = "deep",  source_types = ["academic", "news", "social"],  target_length = "3000_words"}timeout = 900# Stage 2: Entwurf (Verwendung verschiedener Modelle)[[workflows.tasks]]id = "draft"hand = "researcher"              # Wiederverwendung des Researcher-Hands mit anderem Promptconfig = {  model = "claude-opus-4-8",  instruction = "draft_article",  style = "professional_yet_approachable"}depends_on = ["research"]# Stage 3: Faktenprüfung[[workflows.tasks]]id = "fact_check"hand = "researcher"config = {  model = "claude-haiku-4-5",   # Schnelles Modell für die Prüfung  action = "verify_claims",  strictness = "high"}depends_on = ["draft"]# Stage 4: Bilderstellung[[workflows.tasks]]id = "generate_cover"hand = "clip"config = {  action = "generate_cover_image",  style = "modern_tech",  resolution = "1280x720"}depends_on = ["draft"]           # Unabhängig von der Faktenprüfung, daher parallel möglich# Stage 5: SEO-Optimierung[[workflows.tasks]]id = "seo_optimize"hand = "researcher"config = { action = "seo_optimize", target_keyword = "{{.keyword}}" }depends_on = ["fact_check"]# Stage 6: Veröffentlichung[[workflows.tasks]]id = "publish"channel = "webhook"config = { url = "{{.cms_endpoint}}", method = "POST" }depends_on = ["seo_optimize", "generate_cover"]# Stage 7: Promotion[[workflows.tasks]]id = "promote"channel = "twitter"config = { action = "post_thread", hashtags = ["#AI", "#Automation"] }depends_on = ["publish"]

Beachten Sie, dass generate_cover in Stage 4 nur von draft abhängt und nicht von fact_check – das bedeutet, dass die Bilderstellung und die Faktenprüfung parallel ausgeführt werden, was die Wartezeit verkürzt.

Beispiel 4: Kundensupport-Klassifizierung (Bedingtes Routing)

Dies ist ein intelligenter Support-Workflow, der Anfragen basierend auf dem Problemtyp automatisch an verschiedene Bearbeitungsprozesse weiterleitet:

[[workflows]]name = "support_triage"description = "Intelligenter Support: Klassifizierung → Routing → Bearbeitung → Eskalation"[workflows.trigger]type = "message"channel = "whatsapp"pattern = "/help *"# Klassifizierung[[workflows.tasks]]id = "classify"hand = "researcher"config = {  model = "claude-haiku-4-5",  action = "classify_intent",  categories = ["billing", "technical", "feature_request", "complaint"]}# Abrechnung → Abfragesystem[[workflows.tasks]]id = "handle_billing"channel = "webhook"config = { url = "https://billing-api.company.com/lookup" }depends_on = ["classify"]condition = "ctx.intent == 'billing'"# Technik → Wissensdatenbank-Suche[[workflows.tasks]]id = "search_kb"hand = "researcher"config = { action = "search_knowledge_base", max_results = 3 }depends_on = ["classify"]condition = "ctx.intent == 'technical'"# Feature-Anfrage → Product Board Eintrag[[workflows.tasks]]id = "log_feature"channel = "webhook"config = { url = "https://productboard-api.company.com/notes" }depends_on = ["classify"]condition = "ctx.intent == 'feature_request'"# Beschwerde → Sofortige Eskalation an Menschen[[workflows.tasks]]id = "escalate"channel = "slack"config = { channel = "#urgent-support", mention = "@oncall" }depends_on = ["classify"]condition = "ctx.intent == 'complaint'"# Letzte Antwort[[workflows.tasks]]id = "reply"channel = "whatsapp"config = { reply_to = "{{.original_message.id}}" }depends_on = ["handle_billing", "search_kb", "log_feature", "escalate"]

Workflow-Debugging und Tests

Vor der Bereitstellung in der Produktion empfiehlt es sich, den Dry-Run-Modus zu verwenden:

# Dry run (führt keine externen Aktionen aus)openfang workflow run competitor_monitor --dry-run# Nur bis zu einem bestimmten Task ausführenopenfang workflow run content_pipeline --until draft# Ausführungshistorie anzeigenopenfang workflow history competitor_monitor --limit 10# Detaillierte Logs einer bestimmten Ausführung anzeigenopenfang workflow inspect competitor_monitor --run-id abc123

Best Practices für Workflows

Angemessene Timeouts setzen: Jeder Task sollte ein timeout haben, um zu verhindern, dass ein einzelner Schritt den gesamten Prozess blockiert.
Wiederholungsmechanismen hinzufügen: Konfigurieren Sie für Netzwerk-Tasks retry = 3 mit exponentiellem Backoff.
Bedingungen nutzen: Vermeiden Sie unnötige Task-Ausführungen, um API-Kosten zu sparen.
Ausführungsprotokolle: Konfigurieren Sie log_level = "debug" für die Fehlerbehebung.
Workflow-Gesundheit überwachen: Binden Sie Workflow-Metriken in Prometheus/Grafana ein.

Häufig gestellte Fragen

Wie viele Tasks unterstützt ein Workflow maximal?

Es gibt keine harte Begrenzung. Wir empfehlen jedoch, die Anzahl der Tasks pro Workflow auf unter 20 zu halten, um die Wartbarkeit zu gewährleisten. Komplexere Prozesse sollten in mehrere Unter-Workflows aufgeteilt werden.

Wie gehe ich mit fehlgeschlagenen Tasks um?

Jeder Task kann eine individuelle Fehlerstrategie erhalten:

``toml [workflows.tasks.fallback] on_failure = "continue" # continue / retry / abort / escalate retry_count = 3 retry_delay_seconds = 60 fallback_task = "manual_review" # Ersatz-Task bei Fehler``

Können sich Workflows gegenseitig aufrufen?

Ja. Mit trigger.type = "workflow" kann ein Workflow einen anderen auslösen. In Kombination mit Bedingungen lassen sich komplexe verschachtelte Logiken realisieren.

Wie übergebe ich dynamische Parameter in Workflows?

Verwenden Sie die Template-Syntax {{.variable_name}}. Kontextvariablen stammen aus den Trigger-Daten, dem Output vorheriger Tasks oder globalen Konfigurationsvariablen.

Nächste Schritte

OpenFang Hands Konfigurationsleitfaden: Erfahren Sie mehr über die detaillierte Konfiguration jedes Hands.
OpenFang Sicherheits-Best-Practices: Schützen Sie Ihre automatisierten Workflows.