AI-Workflow

AI Workflow ist ein System, das mehrere KI-Modellaufrufe, Tool-Nutzung und Datenverarbeitungsschritte geordnet in einer automatisierten Pipeline kombiniert.

Ein einzelner LLM-Aufruf kann Fragen beantworten, aber reale Aufgaben erfordern oft: das Web durchsuchen → Informationen extrahieren → analysieren → einen Bericht schreiben → eine E-Mail senden. Bei AI Workflow geht es darum, diese Schritte zu verketten, damit die KI die gesamte Aufgabe automatisch erledigen kann, statt nur einen einzelnen Satz zu beantworten.

Eine intuitive Analogie

Stellen Sie sich eine Montagelinie vor:

Mode	Analogie	Beschreibung
Einzelner LLM-Aufruf	Wie ein Handwerker	Geben Sie ihm ein Stück Eisen, und er gibt Ihnen ein Schwert zurück
AI-Workflow	Wie eine ganze Montagelinie	Rohstoffe gehen hinein, durchlaufen automatisch Schmelzen → Schmieden → Abschrecken → Polieren → Verpacken, und fertige Produkte kommen heraus

Jeder Schritt kann ein KI-Modell, eine Codefunktion, eine externe API oder ein menschlicher Prüfknoten sein.

Evolution von Frage und Antwort zu Aktion

Warum AI Workflow benötigt wird

Einschränkungen einzelner Aufrufe

Ein einzelner LLM-Aufruf kann sehr wenig erreichen:

Begrenztes Kontextfenster: kann nicht ein ganzes Buch auf einmal lesen
Kein Zugriff auf Echtzeitinformationen: Trainingsdaten haben einen Stichtag
Kann keine Aktionen ausführen: kann nicht tatsächlich E-Mails senden, Code schreiben und ausführen
Kann sich nicht selbst validieren: kann Fehler nach ihrer Generierung nicht erkennen und korrigieren
Komplexe Aufgaben sind fehleranfällig: zu viel in einem Schritt zu tun führt zu Qualitätsverschlechterung

Fünf Kernprobleme, die AI Workflow löst

Kernkomponenten

Ein vollständiger AI Workflow besteht aus den folgenden Kernelementen.

Detaillierte Erklärung jeder Komponente

LLM (Large Language Model)- Das "Gehirn" des Workflows, verantwortlich für Reasoning, Generierung und Entscheidungsfindung. Häufig verwendet: GPT-4o, Claude 3.5, Gemini 1.5, lokales Llama 3.

Werkzeuge- Schnittstellen, die es KI ermöglichen, mit der Außenwelt zu interagieren, einschließlich: Suchmaschinen (Tavily, Serper, Bing), Code-Ausführungsumgebungen (Python REPL, Sandbox-Umgebungen), Datenbankabfragen (SQL, Vector DB), externe APIs (Wetter, Aktien, E-Mail, Kalender), Dateioperationen (Lesen/Schreiben, PDF/Excel parsen).

Gedächtnis- Kurzzeitgedächtnis speichert den Gesprächsverlauf der aktuellen Sitzung (im Prompt gespeichert); Langzeitgedächtnis erreicht sitzungsübergreifende persistente Speicherung über Vektordatenbank + RAG; Arbeitsgedächtnis hält den Zwischenzustand während der Aufgabenausführung aufrecht.

Zustand- Der Informationsträger, der zwischen Schritten einer Aufgabe weitergegeben wird, wie ein Staffelstab; jeder Schritt kann das Ergebnis des vorherigen Schritts lesen und neue Ergebnisse schreiben.

Router / Bedingung (Router)- Bestimmt anhand der Ausgabe des vorherigen Schritts dynamisch den nächsten Schritt und ermöglicht komplexe Ablaufsteuerung wie Verzweigung, Schleife und Sprung.

Mensch in der Schleife- Pausiert an Schlüsselknoten, um auf menschliche Bestätigung zu warten; geeignet für Hochrisiko-Operationen (z. B. Löschen von Daten, Senden von E-Mails, Finanzoperationen).

Sechs häufige Workflow-Muster

Im Folgenden werden die sechs häufigsten Entwurfsmuster in AI-Workflow vorgestellt, von einfach bis komplex, geeignet für unterschiedliche Szenarien.

Muster 1: Sequenzielle Kette

Das grundlegendste Muster, bei dem die Schritte A → B → C linear ausgeführt werden und die Ausgabe des vorherigen Schritts die Eingabe des nächsten ist.

Dimension	Beschreibung
Anwendbare Szenarien	Pipelines zur Dokumentverarbeitung, Inhaltsgenerierung, Datentransformation
Vorteile	Einfach, vorhersehbar
Nachteile	Starr, kann nicht dynamisch anhand des Inhalts angepasst werden

Muster 2: Bedingtes Routing

Wählt dynamisch unterschiedliche nachfolgende Pfade basierend auf dem Ausgabeinhalt eines bestimmten Schritts aus.

Dimension	Beschreibung
Anwendbare Szenarien	Intelligenter Kundenservice, multifunktionale Assistenten, Klassifizierung und Verarbeitung von Fragen
Vorteile	Flexibel, hohe Ressourcennutzung
Nachteile	Die Routing-Logik erfordert sorgfältiges Design; Klassifizierungsfehler wirken sich auf den gesamten Prozess aus

Muster 3: Parallele Ausführung

Mehrere Teilaufgaben laufen gleichzeitig, und die Ergebnisse werden am Ende zusammengeführt.

Dimension	Beschreibung
Anwendbare Szenarien	Mehrdimensionale Analyse, Batch-Verarbeitung, unabhängige Teilaufgaben
Vorteile	Deutlich verbesserte Geschwindigkeit
Nachteile	Erfordert den Umgang mit Nebenläufigkeitskontrolle und Logik zum Zusammenführen von Ergebnissen

Muster 4: ReAct-Schleife (Reason + Act)

Die KI überlegt zuerst, was zu tun ist, handelt dann, indem sie Tools aufruft, und überlegt weiter basierend auf den Ergebnissen, in einer Schleife, bis die Aufgabe abgeschlossen ist. Dies ist das Kernmuster von AI Agents.

Dimension	Beschreibung
Anwendbare Szenarien	AI Agent, Ausführung komplexer Aufgaben, offene Problemlösung
Vorteile	Dynamisch und flexibel, kann unbekannte Situationen bewältigen
Nachteile	Die Anzahl der Schleifendurchläufe ist nicht kontrollierbar, und es muss eine maximale Schrittzahl festgelegt werden, um eine Endlosschleife zu verhindern.

Muster 5: Planen & Ausführen

Lassen Sie zuerst das LLM einen vollständigen Plan erstellen, und führen Sie dann Schritt für Schritt gemäß dem Plan aus. Der Unterschied zu ReAct ist 'erst klar denken, dann handeln'.

Muster 6: Multi-Agent-Zusammenarbeit

Mehrere spezialisierte Agenten arbeiten zusammen, und jeder Agent hat seine eigene Rolle und sein eigenes Werkzeugset.

Dimension	Beschreibung
Anwendbare Szenarien	Komplexe Softwareentwicklung, Unterstützung bei der Forschung, Unternehmensautomatisierung
Vorteile	Gezielte Spezialisierung, höhere Qualität, einfache Skalierung
Nachteile	Hohe Systemkomplexität; die Kommunikation zwischen Agenten erfordert sorgfältiges Design

Vergleich gängiger Frameworks und Tools

Im Folgenden finden Sie einen umfassenden Vergleich der gängigsten AI-Workflow-Frameworks, damit Sie je nach Ihrer eigenen Situation auswählen können.

Entscheidungsbaum zur Framework-Auswahl

Wählen Sie anhand Ihrer spezifischen Situation mithilfe des folgenden Entscheidungsbaums das passende Framework aus:

你的情况是什么？
│
├─── 没有编程基础，想用可视化工具搭建
│    ├─── 主要是 AI 应用（问答、生成）→ Dify（首选）
│    └─── 需要连接 Slack/邮件等 SaaS 系统 → n8n
│
├─── 有 Python 基础，代码优先
│    ├─── 做知识库 / RAG 系统 → LlamaIndex
│    ├─── 做多 Agent 协作，想快速上手 → CrewAI
│    ├─── 需要复杂有状态流程控制 → LangGraph
│    └─── 通用场景，想要最大生态 → LangChain
│
└─── 已有明确场景，生产级要求
     ├─── 高并发、精细控制 → LangGraph + LangSmith
     └─── 企业部署、私有化 → Dify 自托管

Schnellstart: Python-Codebeispiele

Die folgenden Beispiele reichen von der einfachsten sequenziellen Kette bis zur komplexen Multi-Agent-Zusammenarbeit und zeigen, wie KI-Workflows Schritt für Schritt implementiert werden.

LangChain sequenzielle Kette

Das grundlegendste Workflow-Muster: Mehrere LLM-Aufrufschritte mit dem Pipe-Operator | verketten.

Abhängigkeitspakete installieren:

pip install langchain langchain-openai

Beispiel

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o-mini", api_key="your-api-key")

# ─── Drei Schritte definieren ────────────────────────────────────────────
# Schritt 1: Den Artikel ins Chinesische übersetzen
translate_prompt = ChatPromptTemplate.from_template(
Übersetze den folgenden englischen Artikel ins Chinesische und bewahre die ursprüngliche Bedeutung:\n\n{article}"
)

# Schritt 2: Zusammenfassung extrahieren
summarize_prompt = ChatPromptTemplate.from_template(
Bitte fasse den folgenden Artikel in 3 Kernpunkten zusammen, jeweils ein Satz:\n\n{translated}"
)

# Schritt 3: Titel generieren
title_prompt = ChatPromptTemplate.from_template(
Erzeuge basierend auf der folgenden Zusammenfassung einen attraktiven chinesischen Titel (innerhalb von 15 Zeichen):\n\n{summary}"
)

parser = StrOutputParser()

# ─── Mit | Operator zu einer Pipeline verketten ─────────────────────────────────
chain = (
{"translated": translate_prompt | llm | parser}
| {"summary": summarize_prompt | llm | parser,
"translated": lambda x: x["translated"]}
| title_prompt | llm | parser
)

# ─── Ausführen ────────────────────────────────────────────────────
article = """
Künstliche Intelligenz verändert, wie wir arbeiten und leben.
Von der Automatisierung repetitiver Aufgaben bis zur Unterstützung kreativer Arbeit,
KI-Tools werden in modernen Workflows unverzichtbar...
"""

result = chain.invoke({"article": article})
print(result)

AI 正在重塑现代工作流：从自动化到创意辅助

Tool-Calling-Agent (ReAct-Modus)

ReAct ist der Kernmodus eines KI-Agents; er ermöglicht der KI, zwischen Denken und Handeln zu wechseln, bis die Aufgabe abgeschlossen ist.

Beispiel

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import tool
from langchain import hub
import requests, datetime

# ─── Tools definieren ────────────────────────────────────────────────
@tool
def get_weather(city: str) -> str:
"""Ruft die aktuellen Wetterinformationen für eine angegebene Stadt ab"""
# In Produktion durch eine echte Wetter-API ersetzen
mock_data = {
"Beijing": "Sonnig, 22°C, leichte Brise",
"Shanghai": "Bewölkt, 26°C, Luftfeuchtigkeit 75%",
"Guangzhou": "Leichter Regen, 30°C, nimm einen Regenschirm mit",
}
return mock_data.get(city, f"Keine Wetterdaten für {city} verfügbar")

@tool
def search_web(query: str) -> str:
"""Sucht im Web nach Informationen und gibt eine Zusammenfassung relevanter Inhalte zurück"""
# In Produktion Tavily-/Serper-API integrieren
return f"Suchergebnisse für '{query}': Dies ist ein simuliertes Suchergebnis..."

@tool
def calculate(expression: str) -> str:
"""Berechnet einen mathematischen Ausdruck, z. B. '2 + 3 * 4'"""
try:
result = eval(expression, {"__builtins__": {}}, {})
return str(result)
except Exception as e:
return f"Berechnungsfehler: {e}"

@tool
def get_date() -> str:
"""Gibt das heutige Datum zurück"""
return datetime.date.today().strftime("%Y-%m-%d")

# ─── Agent erstellen ───────────────────────────────────────────────
tools = [get_weather, search_web, calculate, get_date]
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Verwende den Standard-ReAct-Prompt aus dem LangChain Hub
prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True, # Gib jeden Denkschritt aus, um das Debuggen zu erleichtern
max_iterations=8, # Verhindere Endlosschleifen
handle_parsing_errors=True
)

# ─── Ausführen ────────────────────────────────────────────────────
result = agent_executor.invoke({
"input": "Wie ist das heutige Datum? Wie ist das Wetter in Peking? Wenn du draußen 5km zu Fuß gehst, verbrennst du etwa 300 Kalorien, "
"Das Laufen der gleichen Strecke verbrennt etwa 1.6-mal so viele Kalorien wie das Gehen. Bitte berechne die beim Laufen verbrannten Kalorien."
})

print(result["output"])
# Der Agent entscheidet automatisch: zuerst get_date → get_weather("Beijing") → calculate("300*1.6")
# Kombiniere schließlich alle Informationen, um eine vollständige Antwort zu geben

Zustandsbehafteter LangGraph-Workflow

LangGraph definiert komplexe Workflows mithilfe von Graphen, wobei jeder Knoten eine Funktion ist und Kanten die Übergangslogik definieren.

Beispiel

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated
import operator

# ─── Zustandsstruktur definieren (zwischen Knoten übergebene Daten) ────────────────────────
class ResearchState(TypedDict):
topic: str # Forschungsthema
research_notes: str # Recherchenotizen
draft: str # Entwurf
review_feedback: str # Überprüfungsfeedback
final_report: str # Abschlussbericht
revision_count: Annotated[int, operator.add] # Anzahl der Überarbeitungen (kumulativ)

llm = ChatOpenAI(model="gpt-4o")

# ─── Knotenfunktionen definieren ─────────────────────────────────────────────
def research_node(state: ResearchState) -> dict:
"""Knoten 1: Forschungsphase"""
response = llm.invoke(
f"Bitte führe eine kurze Recherche zum folgenden Thema durch und liste 5 Kernpunkte auf: {state['topic']}"
)
return {"research_notes": response.content}

def write_node(state: ResearchState) -> dict:
"""Knoten 2: Entwurf schreiben"""
prompt = f"""
Thema: {state['topic']}
Recherchenotizen: {state['research_notes']}
{'Vorheriges Überprüfungsfeedback: ' + state.get('review_feedback', '') if state.get('review_feedback') else ''}

Bitte schreibe einen 300-Wörter-Entwurf eines Analyseberichts basierend auf dem obigen Inhalt.
"""
response = llm.invoke(prompt)
return {"draft": response.content, "revision_count": 1}

def review_node(state: ResearchState) -> dict:
"""Knoten 3: Entwurf überprüfen"""
response = llm.invoke(
f"Überprüfe den folgenden Bericht. Wenn die Qualität den Standard erfüllt, antworte mit 'APPROVED'; andernfalls gib konkrete Überarbeitungsvorschläge:\n\n{state['draft']}"
)
return {"review_feedback": response.content}

def finalize_node(state: ResearchState) -> dict:
"""Knoten 4: Abschließen"""
return {"final_report": state["draft"]}

# ─── Routing-Funktion: Bestimme, welcher Pfad nach der Überprüfung genommen wird ─────────────────────────────
def should_revise(state: ResearchState) -> str:
if "APPROVED" in state["review_feedback"]:
return "finalize" # → Abschließen
elif state["revision_count"] >= 3:
return "finalize" # → Überschreitet 3 Überarbeitungen, erzwinge Abbruch
else:
return "revise" # → Zurück zum Schreibknoten zur Überarbeitung

# ─── Workflow-Graph erstellen ─────────────────────────────────────────────
workflow = StateGraph(ResearchState)

# Knoten hinzufügen
workflow.add_node("research", research_node)
workflow.add_node("write", write_node)
workflow.add_node("review", review_node)
workflow.add_node("finalize", finalize_node)

# Einstiegspunkt festlegen
workflow.set_entry_point("research")

# Kanten hinzufügen (Übergangslogik definieren)
workflow.add_edge("research", "write") # Recherche → Schreiben
workflow.add_edge("write", "review") # Schreiben → Überprüfung

# Bedingte Kante: Pfad basierend auf Review-Ergebnis wählen
workflow.add_conditional_edges(
"review",
should_revise,
{
"revise": "write", # Überarbeitung nötig → Zurück zum Schreiben
"finalize": "finalize" # Genehmigt → Finalisieren
}
)

workflow.add_edge("finalize", END)

# Kompilieren und ausführen
app = workflow.compile()

result = app.invoke({"topic": "Die Auswirkungen generativer KI auf die Softwareentwicklungsbranche", "revision_count": 0})
print(result["final_report"])

CrewAI Multi-Agenten-Zusammenarbeit

CrewAI ermöglicht es mehreren Agenten, durch die Definition von Rollen und Aufgaben wie ein Team zusammenzuarbeiten.

Beispiel

from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

# ─── Agent definieren (Rolle) ────────────────────────────────────────
researcher = Agent(
role=Marktforscher,
goal=Umfassende, genaue Marktinformationen zum Zielthema sammeln und organisieren.,
backstory=Du bist ein erfahrener Marktanalyst und geschickt darin, wichtige Erkenntnisse aus großen Informationsmengen zu extrahieren.,
llm=llm,
verbose=True
)

analyst = Agent(
role=Datenanalyst,
goal=Basierend auf den vom Marktforscher bereitgestellten Informationen eine eingehende Analyse durchführen und wertvolle Schlussfolgerungen ziehen.,
backstory=Du bist hervorragend darin, Daten zur Kommunikation einzusetzen, und kannst Trends und Chancen aufdecken, die in Informationen verborgen sind.,
llm=llm,
verbose=True
)

writer = Agent(
role=Experte für Berichterstellung,
goal=Die analytischen Schlussfolgerungen in einen klaren, professionellen und überzeugenden Bericht verfassen.,
backstory=Du hast umfangreiche Erfahrung im Verfassen von Geschäftstexten und kannst komplexe Analysen leicht verständlich machen.,
llm=llm,
verbose=True
)

# ─── Aufgabe definieren ────────────────────────────────────────
research_task = Task(
description=Den aktuellen Status des chinesischen Marktes für neue Energiefahrzeuge recherchieren, einschließlich wichtiger Marken, Marktanteile und Wachstumstrends.,
expected_output=Ein Marktforschungsbericht mit 5 zentralen Datenpunkten, einschließlich Zahlen und konkreten Fakten.,
agent=researcher
)

analysis_task = Task(
description=Basierend auf dem Forschungsbericht die Chancen und Risiken der nächsten 3 Jahre analysieren und Empfehlungen für Investitionsbewertungen geben.,
expected_output=SWOT-Analysetabelle + Investitionsbewertung (starke Empfehlung/Empfehlung/neutral/vorsichtig) + Begründung,
agent=analyst,
context=[research_task] Ausgabe der abhängigen Forschungsaufgabe
)

writing_task = Task(
description=Die Recherche und Analyse in ein professionelles Investment-Briefing mit 500 Wörtern und klarer Formatierung integrieren.,
expected_output=Ein Briefing mit vier Abschnitten: Zusammenfassung, Marktstatus, Chancen und Risiken sowie Investitionsempfehlungen.,
agent=writer,
context=[research_task, analysis_task]
)

# ─── Team zusammenstellen und ausführen ───────────────────────────────────────────
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, writing_task],
process=Process.sequential, # Sequentielle Ausführung (kann auch auf hierarchisch umgestellt werden)
verbose=True
)

result = crew.kickoff()
print(result)

Human-in-the-Loop (manueller Überprüfungsknoten)

LangGraph unterstützt das Anhalten an kritischen Schritten, um vor der Fortsetzung der Ausführung auf eine menschliche Bestätigung zu warten.

Beispiel

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict

class EmailState(TypedDict):
recipient: str
content: str
approved: bool

def draft_email(state: EmailState) -> dict:
E-Mail entwerfen.
content = fSehr geehrte/r {state['recipient']},\n\nDies ist der von der KI entworfene E-Mail-Inhalt...\n\nMit freundlichen Grüßen
return {"content": content}

def send_email(state: EmailState) -> dict:
E-Mail senden (Hochrisiko-Operation, erfordert manuelle Überprüfung vor der Ausführung)
print(fE-Mail wurde an {state['recipient']} gesendet)
return {}

# Routing: Entscheiden, ob anhand des Ergebnisses der manuellen Überprüfung gesendet wird.
def check_approval(state: EmailState) -> str:
return "send" if state.get("approved") else END

workflow = StateGraph(EmailState)
workflow.add_node("draft", draft_email)
workflow.add_node("send", send_email)
workflow.set_entry_point("draft")

# Nach Abschluss des Entwurfs pausieren, auf menschliche Überprüfung warten (interrupt_after)
workflow.add_conditional_edges("draft", check_approval, {"send": "send", END: END})
workflow.add_edge("send", END)

# MemorySaver verwenden, um Unterbrechung und Fortsetzung zu unterstützen
memory = MemorySaver()
app = workflow.compile(
checkpointer=memory,
interrupt_after=["draft"] # Nach dem Entwurfs-Knoten pausieren
)

config = {"configurable": {"thread_id": "email-001"}}

# Erster Durchlauf: Nach Ausführung des Entwurfs pausieren
state = app.invoke({"recipient": "Customer A", "approved": False}, config)
print("Entwurf erstellt, warten auf Überprüfung:")
print(state["content"])

# Nach manueller Überprüfung Status aktualisieren und Ausführung fortsetzen
user_input = input("\nSenden genehmigen? (y/n): ")
if user_input.lower() == "y":
app.update_state(config, {"approved": True})
app.invoke(None, config) # Vom Haltepunkt fortsetzen
print("E-Mail gesendet")
else:
print("Senden abgebrochen")

Prompt-Engineering: Zentrale Techniken im Workflow

Im KI-Workflow beeinflusst die Qualität des Prompts direkt die Ausgabequalität jedes Knotens.

Beispiel

# Vage Rollendefinition
# system_prompt = "Du bist ein KI-Assistent, hilf mir, dieses Dokument zu analysieren"

# Klare Rollendefinition (empfohlen)
system_prompt = """
Du bist ein professioneller Finanzanalyst und konzentrierst dich auf die Identifizierung von Risikosignalen in Finanzberichten.
Deine Aufgabe: Extrahiere alle wichtigen Datenpunkte in Bezug auf Verbindlichkeiten, Cashflow und Rentabilität aus dem folgenden Bericht.
Ausgabeformat: JSON, das risk_level (high/medium/low) und key_findings (list) enthält.
Hinweis: Gib nur JSON aus, füge keinen erklärenden Text hinzu.
"""

Beispiel

# Unstrukturierten Text übergeben
# state["previous_output"] = "Analyseergebnis: Dieses Unternehmen sieht gut aus, es gibt einige Risiken..."

# Strukturierte Daten übergeben (empfohlen)
state["analysis_result"] = {
"risk_level": "mittel",
"key_findings": ["Schuldenquote 45%, Branchendurchschnitt 38%", "Cashflow ist positiv, Q3 gegenüber dem Vorquartal um 12% gesunken"],
"recommendation": "Vorsichtig halten"
}

Beispiel

# JSON-Ausgabe im Prompt erzwingen
prompt = """
Bitte analysiere die Stimmung des folgenden Textes.
Gib streng im folgenden JSON-Format aus, füge keinen anderen Inhalt hinzu:
{
"sentiment": "positiv" | "negativ" | "neutral",
"confidence": 0.0-1.0,
"reason": "kurze Erklärung"
}

Text: {text}
"""

Fehlerbehandlung und Wiederholungsversuche

Beispiel

from tenacity import retry, stop_after_attempt, wait_exponential
import logging

@retry(
stop=stop_after_attempt(3), # Bis zu 3-mal wiederholen
wait=wait_exponential(multiplier=1, min=2, max=10) # Exponentielles Backoff-Warten
)
def call_llm_with_retry(prompt: str) -> str:
try:
response = llm.invoke(prompt)
return response.content
except Exception as e:
logging.error(f"LLM-Aufruf fehlgeschlagen: {e}")
raise

def safe_parse_json(text: str) -> dict:
"""JSON-Ausgabe von LLM sicher parsen"""
import json, re
# JSON-Teil extrahieren (LLM fügt manchmal zusätzlichen Erklärungstext hinzu)
json_match = re.search(r'\{.*\}', text, re.DOTALL)
if json_match:
try:
return json.loads(json_match.group())
except json.JSONDecodeError:
pass
return {"error": Parsen fehlgeschlagen, "raw": text}

Kostenkontrolle

Beispiel

# ─── Modelle anhand der Aufgabenwichtigkeit auswählen ────────────────────────────────
def get_llm(task_type: str):
if task_type == "classification": # Für Klassifikationsaufgaben genügt ein leichtgewichtiges Modell
return ChatOpenAI(model="gpt-4o-mini")
elif task_type == "generation": # Für Generierungsaufgaben ein mittleres Modell verwenden
return ChatOpenAI(model="gpt-4o-mini")
elif task_type == "reasoning": # Für komplexes Reasoning ein leistungsstarkes Modell verwenden
return ChatOpenAI(model="gpt-4o")

# ─── Wiederholte Anfragen zwischenspeichern ──────────────────────────────────────────
from langchain.globals import set_llm_cache
from langchain_community.cache import InMemoryCache
set_llm_cache(InMemoryCache()) # Bei identischer Eingabe das zwischengespeicherte Ergebnis direkt zurückgeben; keine doppelte Abrechnung

# ─── Text aufteilen, um übermäßige Länge zu vermeiden ──────────────────────────────────────
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
chunk_size=2000, # 2000 Zeichen pro Chunk
chunk_overlap=200 # 200 Zeichen überlappen, um Kontextkohärenz zu erhalten
)
chunks = splitter.split_text(long_document)

Typische Anwendungsszenarien

KI-Workflow wurde in vielen Bereichen weit verbreitet eingesetzt. Im Folgenden werden die sechs repräsentativsten Szenarien vorgestellt.

Best Practices und häufige Fallstricke

Häufige Fallstricke und Lösungen

Im Folgenden werden die häufigsten Probleme beschrieben, auf die Anfänger bei der Verwendung von KI-Workflow stoßen.

Fallstrick	Symptom	Lösung
Halluzinationskaskade	Die vorherige KI-Ausgabe ist falsch, wird als Tatsache an den nächsten Schritt weitergegeben und verstärkt den Fehler.	Fügen Sie an Schlüsselknoten Verifikationsschritte hinzu; verwenden Sie Tools für numerische Informationen.
Endlosschleife	ReAct-Agent ruft wiederholt Tools auf, ohne zu wissen, wann er aufhören soll.	max_iterations festlegen; dem LLM explizite Abbruchbedingungen geben.
Kontext-Explosion	Mit zunehmenden Schritten wird der an das LLM übergebene Text länger und überschreitet das Fenster.	Übergeben Sie bei jedem Schritt nur die notwendigen Felder; komprimieren Sie historische Informationen mit Zusammenfassungen.
Tool-Missbrauch	Der Agent kann direkt antworten, ruft aber weiterhin Tools auf.	Tool-Beschreibungen optimieren; dem LLM klar sagen, wann Tools nicht benötigt werden.
JSON-Parsing-Fehler	Das Ausgabeformat des LLM ist instabil, was zum Absturz des Programms führt.	safe_parse_json hinzufügen; LangChain OutputParser verwenden.
Kostenüberschreitung	Der Token-Verbrauch wird nicht geschätzt, und die monatliche Rechnung übersteigt die Erwartungen.	Zuerst mit gpt-4o-mini testen; LangSmith zur Überwachung der Nutzung verwenden.
Nebenläufigkeitskonflikt	Mehrere Agenten, die gleichzeitig in denselben Zustand schreiben, verursachen Datenkorruption.	Verwenden Sie die integrierte Zustandsverwaltung von LangGraph; vermeiden Sie die gemeinsame Nutzung von veränderlichem Zustand.

Zusammenfassung und Lernpfad

Wiederholung der Kernkenntnisse

Empfohlener Lernpfad

Phase 1: Grundlagen verstehen (1 Woche)

LLM-API-Aufrufe verstehen, in der Lage sein, Anfragen mit dem OpenAI SDK zu stellen
Das Beispiel der sequenziellen Kette von LangChain in diesem Artikel durcharbeiten
Die Grundprinzipien des Prompt Engineering verstehen

Phase 2: Tools und Agenten (2 Wochen)

Lernen, Tools mit dem @tool-Dekorator zu definieren
Das ReAct-Agent-Beispiel durcharbeiten, die ausführlichen Logs beobachten, um die Schleifenlogik zu verstehen
Dify oder n8n verwenden, um einen visuellen Workflow-Prototyp zu erstellen

Phase 3: Komplexer Workflow (3 Wochen)

LangGraph-Zustandsgraphen lernen, einen Workflow mit bedingter Verzweigung implementieren
CrewAI verwenden, um eine duale Agenten-Kollaborationsaufgabe zu implementieren
Echte Tools integrieren (Tavily Search, Python REPL)
LangSmith verwenden, um Ihren Workflow zu beobachten und zu debuggen

Phase 4: Produktionsreif (kontinuierlich)

Fehler-Wiederholung und strukturiertes Ausgabe-Parsing implementieren
Kostenüberwachung und -optimierung (gestufte Modellnutzung)
In die Produktion bereitstellen, Monitoring und Alerting einrichten

Referenzressourcen

Weitere Erweiterungen

Agenten-Tools

Python-Implementierung von Agenten

AI-Workflow

Eine intuitive Analogie

Evolution von Frage und Antwort zu Aktion

Warum AI Workflow benötigt wird

Einschränkungen einzelner Aufrufe

Fünf Kernprobleme, die AI Workflow löst

Kernkomponenten

Detaillierte Erklärung jeder Komponente

Sechs häufige Workflow-Muster

Muster 1: Sequenzielle Kette

Muster 2: Bedingtes Routing

Muster 3: Parallele Ausführung

Muster 4: ReAct-Schleife (Reason + Act)

Muster 5: Planen & Ausführen

Muster 6: Multi-Agent-Zusammenarbeit

Vergleich gängiger Frameworks und Tools

Entscheidungsbaum zur Framework-Auswahl

Schnellstart: Python-Codebeispiele

LangChain sequenzielle Kette

Beispiel

Tool-Calling-Agent (ReAct-Modus)

Beispiel

Zustandsbehafteter LangGraph-Workflow

Beispiel

CrewAI Multi-Agenten-Zusammenarbeit

Beispiel

Human-in-the-Loop (manueller Überprüfungsknoten)

Beispiel

Prompt-Engineering: Zentrale Techniken im Workflow

Beispiel

Beispiel

Beispiel

Fehlerbehandlung und Wiederholungsversuche

Beispiel

Kostenkontrolle

Beispiel

Typische Anwendungsszenarien

Best Practices und häufige Fallstricke

Häufige Fallstricke und Lösungen

Zusammenfassung und Lernpfad

Wiederholung der Kernkenntnisse

Empfohlener Lernpfad

Referenzressourcen